Initial changes to initialization

Author: Dan Richelson

ldclient/client.py

@@ -35,8 +35,8 @@ class Config(object):
     def __init__(self,
                  base_uri='https://app.launchdarkly.com',
                  events_uri='https://events.launchdarkly.com',
-                 connect_timeout=2,
-                 read_timeout=10,
+                 connect_timeout=10,
+                 read_timeout=15,
                  events_upload_max_batch_size=100,
                  events_max_pending=10000,
                  stream_uri='https://stream.launchdarkly.com',
@@ -149,10 +149,11 @@ def __init__(self, sdk_key, config=None, start_wait=5):
         log.info("Waiting up to " + str(start_wait) + " seconds for LaunchDarkly client to initialize...")
         update_processor_ready.wait(start_wait)
 
-        if self._update_processor.initialized:
+        if self._update_processor.initialized() is True:
             log.info("Started LaunchDarkly Client: OK")
         else:
-            log.info("Initialization timeout exceeded for LaunchDarkly Client. Feature Flags may not yet be available.")
+            log.warn("Initialization timeout exceeded for LaunchDarkly Client or an error occurred. "
+                     "Feature Flags may not yet be available.")
 
     @property
     def sdk_key(self):
@@ -215,7 +216,7 @@ def send_event(value, version=None):
                               'user': user, 'value': value, 'default': default, 'version': version})
 
         if not self.is_initialized():
-            log.warn("Feature Flag evaluation attempted before client has finished initializing! Returning default: "
+            log.warn("Feature Flag evaluation attempted before client has initialized! Returning default: "
                      + str(default) + " for feature key: " + key)
             send_event(default)
             return default

ldclient/polling.py

@@ -23,15 +23,15 @@ def run(self):
             while self._running:
                 start_time = time.time()
                 self._store.init(self._requester.get_all())
-                if not self._ready.is_set() and self._store.initialized:
+                if not self._ready.is_set() is True and self._store.initialized is True:
                     log.info("PollingUpdateProcessor initialized ok")
                     self._ready.set()
                 elapsed = time.time() - start_time
                 if elapsed < self._config.poll_interval:
                     time.sleep(self._config.poll_interval - elapsed)
 
     def initialized(self):
-        return self._running and self._ready.is_set() and self._store.initialized
+        return self._running and self._ready.is_set() is True and self._store.initialized is True
 
     def stop(self):
         log.info("Stopping PollingUpdateProcessor")

ldclient/streaming.py

@@ -1,9 +1,12 @@
+from __future__ import absolute_import
+
 import json
 from threading import Thread
 
 import time
-from sseclient import SSEClient
 
+from requests import HTTPError
+from sseclient import SSEClient
 from ldclient.interfaces import UpdateProcessor
 from ldclient.util import _stream_headers, log
 
@@ -30,8 +33,17 @@ def run(self):
                 for msg in messages:
                     if not self._running:
                         break
-                    if self.process_message(self._store, self._requester, msg, self._ready) is True:
+                    message_ok = self.process_message(self._store, self._requester, msg, self._ready)
+                    if message_ok is True and self._ready.is_set() is False:
                         self._ready.set()
+            except HTTPError as e:
+                if e.response is not None and e.response.status_code is not None:
+                    if 400 <= e.response.status_code < 500:
+                        log.error("StreamingUpdateProcessor response: " + str(e) + ". Retries will not be attempted.")
+                        if self._ready.is_set() is False:
+                            self._ready.set()
+                        self._running = False
+                        return
             except Exception as e:
                 log.error("Could not connect to LaunchDarkly stream: " + str(e.message) +
                           " waiting 1 second before trying again.")
@@ -42,15 +54,15 @@ def stop(self):
         self._running = False
 
     def initialized(self):
-        return self._running and self._ready.is_set() and self._store.initialized
+        return self._running and self._ready.is_set() is True and self._store.initialized is True
 
     @staticmethod
     def process_message(store, requester, msg, ready):
         log.debug("Received stream event {} with data: {}".format(msg.event, msg.data))
         if msg.event == 'put':
             payload = json.loads(msg.data)
             store.init(payload)
-            if not ready.is_set() and store.initialized:
+            if not ready.is_set() is True and store.initialized is True:
                 log.info("StreamingUpdateProcessor initialized ok")
                 return True
         elif msg.event == 'patch':
@@ -63,7 +75,7 @@ def process_message(store, requester, msg, ready):
             store.upsert(key, requester.get_one(key))
         elif msg.event == "indirect/put":
             store.init(requester.get_all())
-            if not ready.is_set() and store.initialized:
+            if not ready.is_set() is True and store.initialized is True:
                 log.info("StreamingUpdateProcessor initialized ok")
                 return True
         elif msg.event == 'delete':