Merge pull request #560 from splitio/rbs_sse

Updated SSE classes

Author: chillaq

splitio/models/rule_based_segments.py

@@ -11,14 +11,14 @@
 class RuleBasedSegment(object):
     """RuleBasedSegment object class."""
 
-    def __init__(self, name, traffic_yype_Name, change_number, status, conditions, excluded):
+    def __init__(self, name, traffic_type_name, change_number, status, conditions, excluded):
         """
         Class constructor.
 
         :param name: Segment name.
         :type name: str
-        :param traffic_yype_Name: traffic type name.
-        :type traffic_yype_Name: str
+        :param traffic_type_name: traffic type name.
+        :type traffic_type_name: str
         :param change_number: change number.
         :type change_number: str
         :param status: status.
@@ -29,7 +29,7 @@ def __init__(self, name, traffic_yype_Name, change_number, status, conditions, e
         :type excluded: Excluded
         """
         self._name = name
-        self._traffic_yype_Name = traffic_yype_Name
+        self._traffic_type_name = traffic_type_name
         self._change_number = change_number
         self._status = status
         self._conditions = conditions
@@ -41,9 +41,9 @@ def name(self):
         return self._name
 
     @property
-    def traffic_yype_Name(self):
+    def traffic_type_name(self):
         """Return traffic type name."""
-        return self._traffic_yype_Name
+        return self._traffic_type_name
 
     @property
     def change_number(self):
@@ -65,6 +65,17 @@ def excluded(self):
         """Return excluded."""
         return self._excluded
 
+    def to_json(self):
+        """Return a JSON representation of this rule based segment."""
+        return {
+            'changeNumber': self.change_number,
+            'trafficTypeName': self.traffic_type_name,
+            'name': self.name,
+            'status': self.status,
+            'conditions': [c.to_json() for c in self.conditions],
+            'excluded': self.excluded.to_json()
+        }
+        
 def from_raw(raw_rule_based_segment):
     """
     Parse a Rule based segment from a JSON portion of splitChanges.
@@ -111,3 +122,10 @@ def get_excluded_keys(self):
     def get_excluded_segments(self):
         """Return excluded segments"""
         return self._segments
+
+    def to_json(self):
+        """Return a JSON representation of this object."""
+        return {
+            'keys': self._keys,
+            'segments': self._segments
+        }

splitio/models/telemetry.py

@@ -140,6 +140,7 @@ class OperationMode(Enum):
 class UpdateFromSSE(Enum):
     """Update from sse constants"""
     SPLIT_UPDATE = 'sp'
+    RBS_UPDATE = 'rbs'
 
 def get_latency_bucket_index(micros):
     """

splitio/push/parser.py

@@ -28,6 +28,7 @@ class UpdateType(Enum):
     SPLIT_UPDATE = 'SPLIT_UPDATE'
     SPLIT_KILL = 'SPLIT_KILL'
     SEGMENT_UPDATE = 'SEGMENT_UPDATE'
+    RB_SEGMENT_UPDATE = 'RB_SEGMENT_UPDATE'
 
 
 class ControlType(Enum):
@@ -329,7 +330,7 @@ def __init__(self, channel, timestamp, change_number, previous_change_number, fe
         """Class constructor."""
         BaseUpdate.__init__(self, channel, timestamp, change_number)
         self._previous_change_number = previous_change_number
-        self._feature_flag_definition = feature_flag_definition
+        self._object_definition = feature_flag_definition
         self._compression = compression
 
     @property
@@ -352,13 +353,13 @@ def previous_change_number(self):  # pylint:disable=no-self-use
         return self._previous_change_number
 
     @property
-    def feature_flag_definition(self):  # pylint:disable=no-self-use
+    def object_definition(self):  # pylint:disable=no-self-use
         """
         Return feature flag definition
         :returns: The new feature flag definition
         :rtype: str
         """
-        return self._feature_flag_definition
+        return self._object_definition
 
     @property
     def compression(self):  # pylint:disable=no-self-use
@@ -451,6 +452,56 @@ def __str__(self):
         """Return string representation."""
         return "SegmentChange - changeNumber=%d, name=%s" % (self.change_number, self.segment_name)
 
+class RBSChangeUpdate(BaseUpdate):
+    """rbs Change notification."""
+
+    def __init__(self, channel, timestamp, change_number, previous_change_number, rbs_definition, compression):
+        """Class constructor."""
+        BaseUpdate.__init__(self, channel, timestamp, change_number)
+        self._previous_change_number = previous_change_number
+        self._object_definition = rbs_definition
+        self._compression = compression
+
+    @property
+    def update_type(self):  # pylint:disable=no-self-use
+        """
+        Return the message type.
+
+        :returns: The type of this parsed Update.
+        :rtype: UpdateType
+        """
+        return UpdateType.RB_SEGMENT_UPDATE
+
+    @property
+    def previous_change_number(self):  # pylint:disable=no-self-use
+        """
+        Return previous change number
+        :returns: The previous change number
+        :rtype: int
+        """
+        return self._previous_change_number
+
+    @property
+    def object_definition(self):  # pylint:disable=no-self-use
+        """
+        Return rbs definition
+        :returns: The new rbs definition
+        :rtype: str
+        """
+        return self._object_definition
+
+    @property
+    def compression(self):  # pylint:disable=no-self-use
+        """
+        Return previous compression type
+        :returns: The compression type
+        :rtype: int
+        """
+        return self._compression
+
+    def __str__(self):
+        """Return string representation."""
+        return "RBSChange - changeNumber=%d" % (self.change_number)
 
 class ControlMessage(BaseMessage):
     """Control notification."""
@@ -503,6 +554,9 @@ def _parse_update(channel, timestamp, data):
     if update_type == UpdateType.SPLIT_UPDATE and change_number is not None:
         return SplitChangeUpdate(channel, timestamp, change_number, data.get('pcn'), data.get('d'), data.get('c'))
 
+    if update_type == UpdateType.RB_SEGMENT_UPDATE and change_number is not None:
+        return RBSChangeUpdate(channel, timestamp, change_number, data.get('pcn'), data.get('d'), data.get('c'))
+
     elif update_type == UpdateType.SPLIT_KILL and change_number is not None:
         return SplitKillUpdate(channel, timestamp, change_number,
                                data['splitName'], data['defaultTreatment'])

splitio/push/processor.py

@@ -35,12 +35,13 @@ def __init__(self, synchronizer, telemetry_runtime_producer):
         self._feature_flag_queue = Queue()
         self._segments_queue = Queue()
         self._synchronizer = synchronizer
-        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, synchronizer.synchronize_segment, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, synchronizer.segment_storage, telemetry_runtime_producer)
+        self._feature_flag_worker = SplitWorker(synchronizer.synchronize_splits, synchronizer.synchronize_segment, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, synchronizer.segment_storage, telemetry_runtime_producer, synchronizer.split_sync.rule_based_segment_storage)
         self._segments_worker = SegmentWorker(synchronizer.synchronize_segment, self._segments_queue)
         self._handlers = {
             UpdateType.SPLIT_UPDATE: self._handle_feature_flag_update,
             UpdateType.SPLIT_KILL: self._handle_feature_flag_kill,
-            UpdateType.SEGMENT_UPDATE: self._handle_segment_change
+            UpdateType.SEGMENT_UPDATE: self._handle_segment_change,
+            UpdateType.RB_SEGMENT_UPDATE: self._handle_feature_flag_update
         }
 
     def _handle_feature_flag_update(self, event):
@@ -119,12 +120,13 @@ def __init__(self, synchronizer, telemetry_runtime_producer):
         self._feature_flag_queue = asyncio.Queue()
         self._segments_queue = asyncio.Queue()
         self._synchronizer = synchronizer
-        self._feature_flag_worker = SplitWorkerAsync(synchronizer.synchronize_splits, synchronizer.synchronize_segment, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, synchronizer.segment_storage, telemetry_runtime_producer)
+        self._feature_flag_worker = SplitWorkerAsync(synchronizer.synchronize_splits, synchronizer.synchronize_segment, self._feature_flag_queue, synchronizer.split_sync.feature_flag_storage, synchronizer.segment_storage, telemetry_runtime_producer, synchronizer.split_sync.rule_based_segment_storage)
         self._segments_worker = SegmentWorkerAsync(synchronizer.synchronize_segment, self._segments_queue)
         self._handlers = {
             UpdateType.SPLIT_UPDATE: self._handle_feature_flag_update,
             UpdateType.SPLIT_KILL: self._handle_feature_flag_kill,
-            UpdateType.SEGMENT_UPDATE: self._handle_segment_change
+            UpdateType.SEGMENT_UPDATE: self._handle_segment_change,
+            UpdateType.RB_SEGMENT_UPDATE: self._handle_feature_flag_update            
         }
 
     async def _handle_feature_flag_update(self, event):