Merge pull request #55 from launchdarkly/dr/twistedv2

Add Twisted support for LDD mode

Author: drichelson

CHANGELOG.md

@@ -2,6 +2,13 @@
 
 All notable changes to the LaunchDarkly Python SDK will be documented in this file. This project adheres to [Semantic Versioning](http://semver.org).
 
+## [3.0.0] - 2016-08-22
+### Added
+- Twisted support for LDD mode only.
+
+### Changed
+- FeatureStore interface get() and all() methods now take an additional callback parameter.
+
 ## [2.0.0] - 2016-08-10
 ### Added
 - Support for multivariate feature flags. `variation` replaces `toggle` and can return a string, number, dict, or boolean value depending on how the flag is defined.

MANIFEST.in

@@ -1,4 +1,5 @@
 include requirements.txt
 include README.txt
 include test-requirements.txt
+include twisted-requirements.txt
 include redis-requirements.txt

README.md

@@ -26,13 +26,38 @@ Your first feature flag
 -----------------------
 
 1. Create a new feature flag on your [dashboard](https://app.launchdarkly.com)
-2. In your application code, use the feature's key to check wthether the flag is on for each user:
+2. In your application code, use the feature's key to check whether the flag is on for each user:
 
         if client.variation("your.flag.key", {"key": "[email protected]"}, False):
             # application code to show the feature
         else:
             # the code to run if the feature is off
 
+Twisted
+-------
+Twisted is supported for LDD mode only. To run in Twisted/LDD mode, 
+
+1. Use this dependency:
+
+	```
+	ldclient-py[twisted]==3.0.0
+	```
+2. Configure the client:
+
+	```
+	feature_store = TwistedRedisFeatureStore(url='YOUR_REDIS_URL', redis_prefix="ldd-restwrapper", expiration=0)
+	ldclient.config.feature_store = feature_store
+	
+	ldclient.config = ldclient.Config(
+	    use_ldd=use_ldd,
+	    event_consumer_class=TwistedEventConsumer,
+	)
+	ldclient.sdk_key = 'YOUR_SDK_KEY'
+	```
+3. Get the client:
+
+	```client = ldclient.get()```
+
 Learn more
 -----------
 

ldclient/client.py

@@ -120,10 +120,7 @@ def __init__(self, sdk_key, config=None, start_wait=5):
             self._event_consumer.start()
 
         if self._config.use_ldd:
-            if self._store.__class__ == "RedisFeatureStore":
-                log.info("Started LaunchDarkly Client in LDD mode")
-                return
-            log.error("LDD mode requires a RedisFeatureStore.")
+            log.info("Started LaunchDarkly Client in LDD mode")
             return
 
         if self._config.feature_requester_class:
@@ -136,6 +133,7 @@ def __init__(self, sdk_key, config=None, start_wait=5):
         update_processor_ready = threading.Event()
 
         if self._config.update_processor_class:
+            log.info("Using user-specified update processor: " + str(self._config.update_processor_class))
             self._update_processor = self._config.update_processor_class(
                 sdk_key, self._config, self._feature_requester, self._store, update_processor_ready)
         else:
@@ -230,23 +228,35 @@ def send_event(value, version=None):
         if user.get('key', "") == "":
             log.warn("User key is blank. Flag evaluation will proceed, but the user will not be stored in LaunchDarkly.")
 
-        flag = self._store.get(key)
-        if not flag:
-            log.warn("Feature Flag key: " + key + " not found in Feature Store. Returning default.")
-            send_event(default)
+        def cb(flag):
+            try:
+                if not flag:
+                    log.warn("Feature Flag key: " + key + " not found in Feature Store. Returning default.")
+                    send_event(default)
+                    return default
+
+                return self._evaluate_and_send_events(flag, user, default)
+
+            except Exception as e:
+                log.error("Exception caught in variation: " + e.message + " for flag key: " + key + " and user: " + str(user))
+
             return default
 
-        value, events = evaluate(flag, user, self._store)
+        return self._store.get(key, cb)
+
+    def _evaluate(self, flag, user):
+        return evaluate(flag, user, self._store)
+
+    def _evaluate_and_send_events(self, flag, user, default):
+        value, events = self._evaluate(flag, user)
         for event in events or []:
             self._send_event(event)
-            log.debug("Sending event: " + str(event))
-
-        if value is not None:
-            send_event(value, flag.get('version'))
-            return value
 
-        send_event(default, flag.get('version'))
-        return default
+        if value is None:
+            value = default
+        self._send_event({'kind': 'feature', 'key': flag.get('key'),
+                          'user': user, 'value': value, 'default': default, 'version': flag.get('version')})
+        return value
 
     def all_flags(self, user):
         if self._config.offline:
@@ -261,7 +271,17 @@ def all_flags(self, user):
             log.warn("User or user key is None when calling all_flags(). Returning None.")
             return None
 
-        return {k: evaluate(v, user, self._store)[0] for k, v in self._store.all().items() or {}}
+        def cb(all_flags):
+            try:
+                return self._evaluate_multi(user, all_flags)
+            except Exception as e:
+                log.error("Exception caught in all_flags: " + e.message + " for user: " + str(user))
+            return {}
+
+        return self._store.all(cb)
+
+    def _evaluate_multi(self, user, flags):
+        return {k: self._evaluate(v, user)[0] for k, v in flags.items() or {}}
 
     def secure_mode_hash(self, user):
         if user.get('key') is None:

ldclient/feature_store.py

@@ -10,24 +10,24 @@ def __init__(self):
         self._initialized = False
         self._features = {}
 
-    def get(self, key):
+    def get(self, key, callback):
         try:
             self._lock.rlock()
             f = self._features.get(key)
             if f is None:
                 log.debug("Attempted to get missing feature: " + str(key) + " Returning None")
-                return None
+                return callback(None)
             if 'deleted' in f and f['deleted']:
                 log.debug("Attempted to get deleted feature: " + str(key) + " Returning None")
-                return None
-            return f
+                return callback(None)
+            return callback(f)
         finally:
             self._lock.runlock()
 
-    def all(self):
+    def all(self, callback):
         try:
             self._lock.rlock()
-            return dict((k, f) for k, f in self._features.items() if ('deleted' not in f) or not f['deleted'])
+            return callback(dict((k, f) for k, f in self._features.items() if ('deleted' not in f) or not f['deleted']))
         finally:
             self._lock.runlock()
 

ldclient/flag.py

@@ -21,18 +21,15 @@ def evaluate(flag, user, store):
         if value is not None:
             return value, prereq_events
 
-    if 'offVariation' in flag and flag['offVariation']:
-        value = _get_variation(flag, flag['offVariation'])
-        return value, prereq_events
-    return None, prereq_events
+    return _get_off_variation(flag), prereq_events
 
 
 def _evaluate(flag, user, store, prereq_events=None):
     events = prereq_events or []
     failed_prereq = None
     prereq_value = None
     for prereq in flag.get('prerequisites') or []:
-        prereq_flag = store.get(prereq.get('key'))
+        prereq_flag = store.get(prereq.get('key'), lambda x: x)
         if prereq_flag is None:
             log.warn("Missing prereq flag: " + prereq.get('key'))
             failed_prereq = prereq

ldclient/interfaces.py

@@ -8,22 +8,23 @@ class FeatureStore(object):
     __metaclass__ = ABCMeta
 
     @abstractmethod
-    def get(self, key):
+    def get(self, key, callback):
         """
-        Gets the data for a feature flag for evaluation
-
-        :param key: The feature flag key
+        Gets a feature and calls the callback with the feature data to return the result
+        :param key: The feature key
         :type key: str
-        :return: The feature flag data
-        :rtype: dict
+        :param callback: The function that accepts the feature data and returns the feature value
+        :type callback: Function that processes the feature flag once received.
+        :return: The result of executing callback.
         """
 
     @abstractmethod
-    def all(self):
+    def all(self, callback):
         """
         Returns all feature flags and their data
-
-        :rtype: dict[str, dict]
+        :param callback: The function that accepts the feature data and returns the feature value
+        :type callback: Function that processes the feature flags once received.
+        :rtype: The result of executing callback.
         """
 
     @abstractmethod

ldclient/redis_feature_store.py

@@ -40,43 +40,44 @@ def init(self, features):
             pipe.hset(self._features_key, k, f_json)
             self._cache[k] = f
         pipe.execute()
+        log.info("Initialized RedisFeatureStore with " + str(len(features)) + " feature flags")
 
-    def all(self):
+    def all(self, callback):
         r = redis.Redis(connection_pool=self._pool)
         all_features = r.hgetall(self._features_key)
         if all_features is None or all_features is "":
             log.warn("RedisFeatureStore: call to get all flags returned no results. Returning None.")
-            return None
+            return callback(None)
 
         results = {}
         for k, f_json in all_features.items() or {}:
             f = json.loads(f_json.decode('utf-8'))
             if 'deleted' in f and f['deleted'] is False:
                 results[f['key']] = f
-        return results
+        return callback(results)
 
-    def get(self, key):
+    def get(self, key, callback=lambda x: x):
         f = self._cache.get(key)
         if f is not None:
             # reset ttl
             self._cache[key] = f
             if f.get('deleted', False) is True:
                 log.warn("RedisFeatureStore: get returned deleted flag from in-memory cache. Returning None.")
-                return None
-            return f
+                return callback(None)
+            return callback(f)
 
         r = redis.Redis(connection_pool=self._pool)
         f_json = r.hget(self._features_key, key)
         if f_json is None or f_json is "":
             log.warn("RedisFeatureStore: feature flag with key: " + key + " not found in Redis. Returning None.")
-            return None
+            return callback(None)
 
         f = json.loads(f_json.decode('utf-8'))
         if f.get('deleted', False) is True:
             log.warn("RedisFeatureStore: get returned deleted flag from Redis. Returning None.")
-            return None
+            return callback(None)
         self._cache[key] = f
-        return f
+        return callback(f)
 
     def delete(self, key, version):
         r = redis.Redis(connection_pool=self._pool)

ldclient/streaming.py

@@ -9,7 +9,6 @@
 
 
 class StreamingUpdateProcessor(Thread, UpdateProcessor):
-
     def __init__(self, sdk_key, config, requester, store, ready):
         Thread.__init__(self)
         self.daemon = True
@@ -31,7 +30,8 @@ def run(self):
                 for msg in messages:
                     if not self._running:
                         break
-                    self.process_message(self._store, self._requester, msg, self._ready)
+                    if self.process_message(self._store, self._requester, msg, self._ready) is True:
+                        self._ready.set()
             except Exception as e:
                 log.error("Could not connect to LaunchDarkly stream: " + str(e.message) +
                           " waiting 1 second before trying again.")
@@ -51,8 +51,8 @@ def process_message(store, requester, msg, ready):
         if msg.event == 'put':
             store.init(payload)
             if not ready.is_set() and store.initialized:
-                ready.set()
                 log.info("StreamingUpdateProcessor initialized ok")
+                return True
         elif msg.event == 'patch':
             key = payload['path'][1:]
             feature = payload['data']
@@ -64,12 +64,13 @@ def process_message(store, requester, msg, ready):
         elif msg.event == "indirect/put":
             store.init(requester.get_all())
             if not ready.is_set() and store.initialized:
-                ready.set()
                 log.info("StreamingUpdateProcessor initialized ok")
+                return True
         elif msg.event == 'delete':
             key = payload['path'][1:]
             # noinspection PyShadowingNames
             version = payload['version']
             store.delete(key, version)
         else:
-            log.warning('Unhandled event in stream processor: ' + msg.event)
+            log.warning('Unhandled event in stream processor: ' + msg.event)
+        return False