Added queue flush interval

Author: amar-jaiswal-haptik

track/client.py

@@ -19,7 +19,8 @@ class Client(object):
     logger = logging.getLogger('interakt')
 
     def __init__(self, write_key=None, host=None, debug=False,
-                 sync_mode=True, timeout=10, max_queue_size=10000, on_error=None, max_retries=3):
+                 sync_mode=True, timeout=10, max_queue_size=10000,
+                 on_error=None, max_retries=3, flush_interval=0.5):
         """Create a new interakt client"""
         require('write_key', write_key, str)
 
@@ -40,7 +41,7 @@ def __init__(self, write_key=None, host=None, debug=False,
             self.consumer = Consumer(
                 queue=self.queue, write_key=write_key,
                 host=host, on_error=on_error, retries=max_retries,
-                timeout=timeout
+                timeout=timeout, flush_interval=flush_interval
             )
             self.consumer.start()
 

track/consumer.py

@@ -14,7 +14,8 @@ class Consumer(Thread):
     logger = logging.getLogger('interakt')
 
     def __init__(self, queue, write_key, host=None,
-                 on_error=None, retries=3, timeout=10):
+                 on_error=None, retries=3, timeout=10,
+                 flush_interval=0.5):
         Thread.__init__(self)
         self.running = True
         self.queue = queue
@@ -23,6 +24,7 @@ def __init__(self, queue, write_key, host=None,
         self.on_error = on_error
         self.retries = retries
         self.timeout = timeout
+        self.flush_interval = flush_interval
 
     def run(self):
         """Runs the consumer."""
@@ -40,7 +42,7 @@ def upload(self):
         queue = self.queue
         queue_msg = None
         try:
-            queue_msg = queue.get(block=True, timeout=1)
+            queue_msg = queue.get(block=True, timeout=self.flush_interval)
         except Empty:
             self.logger.debug("queue is empty now")