streams shouldn't use the same read timeout as the rest of the SDK (#132)

Author: eli-darkly

ldclient/impl/http.py

@@ -17,15 +17,22 @@ def _http_factory(config):
     return HTTPFactory(_base_headers(config), config.http)
 
 class HTTPFactory(object):
-    def __init__(self, base_headers, http_config):
+    def __init__(self, base_headers, http_config, override_read_timeout=None):
         self.__base_headers = base_headers
         self.__http_config = http_config
-        self.__timeout = urllib3.Timeout(connect=http_config.connect_timeout, read=http_config.read_timeout)
+        self.__timeout = urllib3.Timeout(
+            connect=http_config.connect_timeout,
+            read=http_config.read_timeout if override_read_timeout is None else override_read_timeout
+        )
 
     @property
     def base_headers(self):
         return self.__base_headers
 
+    @property
+    def http_config(self):
+        return self.__http_config
+    
     @property
     def timeout(self):
         return self.__timeout

ldclient/sse_client.py

@@ -28,16 +28,14 @@ def __init__(self, url, last_id=None, retry=3000, connect_timeout=10, read_timeo
         self.url = url
         self.last_id = last_id
         self.retry = retry
-        self._connect_timeout = connect_timeout
-        self._read_timeout = read_timeout
         self._chunk_size = chunk_size
 
         if http_factory:
             self._timeout = http_factory.timeout
             base_headers = http_factory.base_headers
         else:
             # for backward compatibility in case anyone else is using this class
-            self._timeout = urllib3.Timeout(connect=self._connect_timeout, read=self._read_timeout)
+            self._timeout = urllib3.Timeout(connect=connect_timeout, read=read_timeout)
             base_headers = {}
 
         # Optional support for passing in an HTTP client

ldclient/streaming.py

@@ -12,7 +12,7 @@
 import math
 import time
 
-from ldclient.impl.http import _http_factory
+from ldclient.impl.http import HTTPFactory, _http_factory
 from ldclient.impl.retry_delay import RetryDelayStrategy, DefaultBackoffStrategy, DefaultJitterStrategy
 from ldclient.interfaces import UpdateProcessor
 from ldclient.sse_client import SSEClient
@@ -75,8 +75,8 @@ def run(self):
                 messages = self._connect()
                 for msg in messages:
                     if not self._running:
-                        log.warning("but I'm done")
                         break
+                    self._retry_delay.set_good_since(time.time())
                     message_ok = self.process_message(self._store, self._requester, msg)
                     if message_ok:
                         self._record_stream_init(False)
@@ -104,10 +104,13 @@ def _record_stream_init(self, failed):
             self._diagnostic_accumulator.record_stream_init(current_time, current_time - self._es_started, failed)
 
     def _connect(self):
+        # We don't want the stream to use the same read timeout as the rest of the SDK.
+        http_factory = _http_factory(self._config)
+        stream_http_factory = HTTPFactory(http_factory.base_headers, http_factory.http_config, override_read_timeout=stream_read_timeout)
         return SSEClient(
             self._uri,
             retry = None,  # we're implementing our own retry
-            http_factory = _http_factory(self._config)
+            http_factory = stream_http_factory
         )
 
     def stop(self):