polishing

Author: chillaq

splitio/push/splitworker.py

@@ -38,19 +38,20 @@ def __init__(self, synchronize_feature_flag, feature_flag_queue, feature_flag_st
         self._running = False
         self._worker = None
         self._feature_flag_storage = feature_flag_storage
+        self._compression_handlers = {
+            CompressionMode.NO_COMPRESSION: lambda event: base64.b64decode(event.feature_flag_definition),
+            CompressionMode.GZIP_COMPRESSION: lambda event: gzip.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8'),
+            CompressionMode.ZLIB_COMPRESSION: lambda event: zlib.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8'),
+        }
 
     def is_running(self):
         """Return whether the working is running."""
         return self._running
 
     def _get_feature_flag_definition(self, event):
         """return feature flag definition in event."""
-        if event.compression == CompressionMode.NO_COMPRESSION.value:
-            return base64.b64decode(event.feature_flag_definition)
-        elif event.compression == CompressionMode.GZIP_COMPRESSION.value:
-            return gzip.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8')
-        elif event.compression == CompressionMode.ZLIB_COMPRESSION.value:
-            return zlib.decompress(base64.b64decode(event.feature_flag_definition)).decode('utf-8')
+        cm = CompressionMode(event.compression) # will throw if the number is not defined in compression mode
+        return self._compression_handlers[cm](event)
 
     def _run(self):
         """Run worker handler."""
@@ -63,10 +64,14 @@ def _run(self):
             _LOGGER.debug('Processing feature flag update %d', event.change_number)
             try:
                 if event.compression is not None and event.previous_change_number == self._feature_flag_storage.get_change_number():
-                    feature_flag_definition = self._get_feature_flag_definition(event)
-                    self._feature_flag_storage.put(from_raw(json.loads(self._get_feature_flag_definition(event))))
-                    self._feature_flag_storage.set_change_number(event.change_number)
-                    continue
+                    try:
+                        self._feature_flag_storage.put(from_raw(json.loads(self._get_feature_flag_definition(event))))
+                        self._feature_flag_storage.set_change_number(event.change_number)
+                        continue
+                    except Exception as e:
+                        _LOGGER.error('Exception raised in updating feature flag')
+                        _LOGGER.error(str(e))
+                        pass
                 self._handler(event.change_number)
             except Exception as e:  # pylint: disable=broad-except
                 _LOGGER.error('Exception raised in feature flag synchronization')

tests/push/test_split_worker.py

@@ -56,15 +56,32 @@ def test_handler(self, mocker):
         def get_change_number():
             return 2345
 
+        self._feature_flag = None
+        def put(feature_flag):
+            self._feature_flag = feature_flag
+
+        self.new_change_number = 0
+        def set_change_number(new_change_number):
+            self.new_change_number = new_change_number
+
         split_worker._feature_flag_storage.get_change_number = get_change_number
+        split_worker._feature_flag_storage.set_change_number = set_change_number
+        split_worker._feature_flag_storage.put = put
+
         # should call the handler
         q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 12345,  "{}", 1))
         time.sleep(0.1)
         assert change_number_received == 123456790
 
+        # should call the handler
+        change_number_received = 0
+        q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456790, 12345,  "{}", 3))
+        time.sleep(0.1)
+        assert change_number_received == 123456790
+
         # should Not call the handler
         change_number_received = 0
-        q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456, 2345, "{}", 1))
+        q.put(SplitChangeUpdate('some', 'SPLIT_UPDATE', 123456, 2345, "eJzEUtFq20AQ/JUwz2c4WZZr3ZupTQh1FKjcQinGrKU95cjpZE6nh9To34ssJ3FNX0sfd3Zm53b2TgietDbF9vXIGdUMha5lDwFTQiGOmTQlchLRPJlEEZeTVJZ6oimWZTpP5WyWQMCNyoOxZPft0ZoA8TZ5aW1TUDCNg4qk/AueM5dQkyiez6IonS6mAu0IzWWSxovFLBZoA4WuhcLy8/bh+xoCL8bagaXJtixQsqbOhq1nCjW7AIVGawgUz+Qqzrr6wB4qmi9m00/JIk7TZCpAtmqgpgJF47SpOn9+UQt16s9YaS71z9NHOYQFha9Pm83Tty0EagrFM/t733RHqIFZH4wb7LDMVh+Ecc4Lv+ZsuQiNH8hXF3hLv39XXNCHbJ+v7x/X2eDmuKLA74sPihVr47jMuRpWfxy1Kwo0GLQjmv1xpBFD3+96gSP5cLVouM7QQaA1vxhK9uKmd853bEZS9jsBSwe2UDDu7mJxd2Mo/muQy81m/2X9I7+N8R/FcPmUd76zjH7X/w4AAP//90glTw==", 2))
         time.sleep(0.1)
         assert change_number_received == 0