bug fixes

Author: davidsulpy

.gitignore

@@ -56,4 +56,6 @@ target/
 .DS_Store
 
 # example app private files 
-isstreamer.ini
+isstreamer.ini
+
+example_app/unpublished/*

CHANGES.txt

@@ -6,7 +6,7 @@
 * added a wait time based on the `Retry-After` header responded by the API when a `429` response is received 
 * using Initial State API >=0.0.4
 
-0.0.26 (unreleased)
+0.0.26
 -------------------
 
 * bubbling up exception for 402 responses from api to be clearer.

ISStreamer/Streamer.py

@@ -131,13 +131,12 @@ def __ship(retry_attempts, wait=0):
 				response = conn.getresponse()
 
 				if (response.status >= 200 and response.status < 300):
-					self.console_message("ship: " + str(response.status), level=2)
-					self.console_message("ship: headers: " + str(response.msg), level=2)
-					self.console_message("ship: body: " + str(response.read()), level=2)
+					self.console_message("ship: status: " + str(response.status) + "\nheaders: " + str(response.msg), level=2)
+					self.console_message("ship: body: " + str(response.read()), level=3)
 				elif (response.status == 401 or response.status == 403):
 					self.console_message("ERROR: unauthorized access_key: " + self.AccessKey)
 				elif (response.status == 402):
-					self.console_message("AccessKey exceeded limit for month, check account at www.initialstate.com/app")
+					self.console_message("AccessKey exceeded limit for month, check account at https://app.initialstate.com/#/account")
 					raise Exception("PAYMENT_REQUIRED")
 				elif (response.status == 429):
 					if "Retry-After" in response.msg:
@@ -203,8 +202,9 @@ def flush(self):
 				self.console_message("flush: queue empty...", level=2)
 		if len(messages) > 0:
 			self.console_message("flush: queue not empty, shipping", level=2)
+			
 			self.ship_messages(messages)
-		self.console_message("flush: finished flushing queue", level=2)
+			self.console_message("flush: finished flushing queue", level=2)
 
 
 	def log_object(self, obj, key_prefix=None, epoch=None):
@@ -265,8 +265,9 @@ def __ship_buffer():
 		if (not self.Offline):
 			if (len(self.LogQueue) >= self.BufferSize):
 				self.console_message("log: queue size approximately at or greater than buffer size, shipping!", level=10)
-				
+				self.console_message("log: async is {state}".format(async=self.Async))
 				if (self.Async):
+					self.console_message("log: spawning ship thread", level=3)
 					t = threading.Thread(target=__ship_buffer)
 					t.daemon = False
 					t.start()

example_app/app.py

@@ -1,15 +1,16 @@
 from ISStreamer.Streamer import Streamer
 import time
 
-streamer = Streamer(bucket_name="test", debug_level=2, ini_file_location="./isstreamer.ini", async=False)
+streamer = Streamer(bucket_name="test", debug_level=3, ini_file_location="./isstreamer.ini", async=True)
 
 def stress_test_loop(i, num):
 	while i > 0:
 		streamer.log("iterations_left_{n}".format(n=num), i)
-		time.sleep(.2)
+		time.sleep(.1)
+		streamer.flush()
 		i = i - 1
 
 
-stress_test_loop(50, 1)
+stress_test_loop(5000, 1)
 
 streamer.close()