Merge 3c13e4a into 4d56acf

hub/kinesis/listener.py

@@ -1,13 +1,13 @@
 import json
 import logging
 import os
-import threading
 import time
 from datetime import datetime, timedelta
 from typing import Callable
 
 from botocore.exceptions import (
     ClientError,
+    ConnectionClosedError,
     ConnectTimeoutError,
     ReadTimeoutError,
 )
@@ -16,17 +16,7 @@
 from hub.kinesis.helpers import create_stream, stream_is_active
 from hub.kinesis.producer import Producer
 
-KINESIS_TIME_SLEEP = int(os.getenv('KINESIS_TIME_SLEEP', '1'))
-
-
-def sleep_listener() -> None:
-    def sleep() -> None:
-        time.sleep(KINESIS_TIME_SLEEP)
-
-    thread = threading.Thread(target=sleep)
-    thread.start()
-    # wait listener to wake up
-    thread.join()
+KINESIS_TIME_SLEEP = float(os.getenv('KINESIS_TIME_SLEEP', '.3'))
 
 
 class Listener:
@@ -69,8 +59,6 @@ def run(self):
                         resp = self.process_func(data)
                         if resp:
                             Producer.put_data(resp, self.stream_name_response)
-                else:
-                    sleep_listener()
 
                 next_iterator = response['NextShardIterator']
                 if self.tries is not None:
@@ -80,8 +68,9 @@ def run(self):
                 ClientError,
                 ConnectTimeoutError,
                 ReadTimeoutError,
+                ConnectionClosedError,
             ):
-                sleep_listener()
+                pass
 
             except client.exceptions.ExpiredIteratorException:
                 next_iterator = client.get_shard_iterator(
@@ -90,3 +79,4 @@ def run(self):
                     ShardIteratorType='AT_TIMESTAMP',
                     Timestamp=datetime.now() - timedelta(seconds=15),
                 )
+            time.sleep(KINESIS_TIME_SLEEP)

hub/version.py

@@ -1 +1 @@
-__version__ = '1.0.19'
+__version__ = '1.0.20'