Synchronous flushing now waits for existing threads to finish

Author: fblundun

snowplow_tracker/consumer.py

@@ -33,6 +33,7 @@
 logger.setLevel(logging.DEBUG)
 
 DEFAULT_MAX_LENGTH = 10
+THREAD_TIMEOUT = 10
 
 new_contract("protocol", lambda x: x == "http" or x == "https")
 
@@ -71,7 +72,7 @@ def __init__(self, endpoint, protocol="http", port=None, method="get", buffer_si
             :param method:      The HTTP request method
             :type  method:      method
             :param buffer_size: The maximum number of queued events before the buffer is flushed. Default is 10.
-            :type  buffer_size: string | None
+            :type  buffer_size: int | None
             :param on_success:  Callback executed after every HTTP request in a flush has status code 200
                                 Gets passed the number of events flushed.
             :type  on_success:  function | None
@@ -97,6 +98,8 @@ def __init__(self, endpoint, protocol="http", port=None, method="get", buffer_si
         self.on_success = on_success
         self.on_failure = on_failure
 
+        self.threads = []
+
     @staticmethod
     @contract
     def as_collector_uri(endpoint, protocol="http", port=None):
@@ -197,16 +200,25 @@ def sync_flush(self):
         """
         logger.debug("Starting synchronous flush...")
         result = Consumer.flush(self)
+        for t in self.threads:
+            t.join(THREAD_TIMEOUT)
         logger.debug("Finished synchrous flush")
         return result
 
+
 class AsyncConsumer(Consumer):
     """
         Uses threads to send HTTP requests asynchronously
     """
     def flush(self):
+        """
+            Removes all dead threads, then creates a new thread which
+            excecutes the flush method of the base Consumer class
+        """
+        self.threads = [t for t in self.threads if t.isAlive()]
         logger.debug("Flushing thread running...")
         t = threading.Thread(target=super(AsyncConsumer, self).flush)
+        self.threads.append(t)
         t.start()
 
 

snowplow_tracker/tracker.py

@@ -366,7 +366,7 @@ def track_unstruct_event(self, event_vendor, event_name, dict_, context=None, ts
         return self.complete_payload(pb)
 
     @contract
-    def flush(self, async=True):
+    def flush(self, async=False):
         """
             Flush the consumer
 
@@ -375,7 +375,8 @@ def flush(self, async=True):
             :rtype:         tracker | int
         """
         if async:
-            return self.out_queue.flush()
+            self.out_queue.flush()
+            return self
         else:
             return self.out_queue.sync_flush()