Merge pull request #610 from splitio/FME-12220-sdk-events-manager

added events manager and events delivery

Author: chillaq

splitio/events/__init__.py

@@ -0,0 +1,25 @@
+"""Base storage interfaces."""
+import abc
+
+class EventsManagerInterface(object, metaclass=abc.ABCMeta):
+    """Events manager interface implemented as an abstract class."""
+
+    @abc.abstractmethod
+    def register(self, sdk_event, event_handler):
+        pass
+    
+    @abc.abstractmethod
+    def unregister(self, sdk_event):
+        pass
+    
+    @abc.abstractmethod
+    def notify_internal_event(self, sdk_internal_event, event_metadata):
+        pass
+    
+
+class EventsDeliveryInterface(object, metaclass=abc.ABCMeta):
+    """Events Delivery interface."""
+
+    @abc.abstractmethod
+    def deliver(self, sdk_event, event_metadata, event_handler):
+        pass

splitio/events/events_delivery.py

@@ -0,0 +1,21 @@
+"""Events Manager."""
+import logging
+
+from splitio.events import EventsDeliveryInterface
+
+_LOGGER = logging.getLogger(__name__)
+
+class EventsDelivery(EventsDeliveryInterface):
+    """Events Manager class."""
+
+    def __init__(self):
+        """
+        Construct Events Manager instance.
+        """
+
+    def deliver(self, sdk_event, event_metadata, event_handler):
+        try:
+            event_handler(event_metadata)
+        except Exception as ex:
+            _LOGGER.error("Exception when calling handler for Sdk Event %s", sdk_event)
+            _LOGGER.error(ex)

splitio/events/events_manager.py

@@ -0,0 +1,152 @@
+"""Events Manager."""
+import threading
+import logging
+from collections import namedtuple
+import pytest
+
+from splitio.events import EventsManagerInterface
+
+_LOGGER = logging.getLogger(__name__)
+
+ValidSdkEvent = namedtuple('ValidSdkEvent', ['sdk_event', 'valid'])
+ActiveSubscriptions = namedtuple('ActiveSubscriptions', ['triggered', 'handler'])
+
+class EventsManager(EventsManagerInterface):
+    """Events Manager class."""
+
+    def __init__(self, events_configurations, events_delivery):
+        """
+        Construct Events Manager instance.
+        """
+        self._active_subscriptions = {}
+        self._internal_events_status = {}
+        self._events_delivery = events_delivery
+        self._manager_config = events_configurations
+        self._lock = threading.RLock()
+        
+    def register(self, sdk_event, event_handler):
+        if self._active_subscriptions.get(sdk_event) != None:
+            return
+        
+        with self._lock:
+            self._active_subscriptions[sdk_event] = ActiveSubscriptions(False, event_handler)
+
+    def unregister(self, sdk_event):
+        if self._active_subscriptions.get(sdk_event) == None:
+            return
+        
+        with self._lock:
+            del self._active_subscriptions[sdk_event]
+
+    def notify_internal_event(self, sdk_internal_event, event_metadata):
+        with self._lock:
+            for sorted_event in self._manager_config.evaluation_order:
+                if sorted_event in self._get_sdk_event_if_applicable(sdk_internal_event):
+                    _LOGGER.debug("EventsManager: Firing Sdk event %s", sorted_event)
+                    if self._get_event_handler(sorted_event) != None:
+                        notify_event = threading.Thread(target=self._events_delivery.deliver, args=[sorted_event, event_metadata, self._get_event_handler(sorted_event)],
+                                                name='SplitSDKEventNotify', daemon=True)
+                        notify_event.start()
+                        self._set_sdk_event_triggered(sorted_event)
+    
+    def _event_already_triggered(self, sdk_event):
+        if self._active_subscriptions.get(sdk_event) != None:
+            return self._active_subscriptions.get(sdk_event).triggered
+        
+        return False
+    
+    def _get_internal_event_status(self, sdk_internal_event):
+        if self._internal_events_status.get(sdk_internal_event) != None:
+            return self._internal_events_status[sdk_internal_event]
+        
+        return False
+    
+    def _update_internal_event_status(self, sdk_internal_event, status):
+        with self._lock:
+            self._internal_events_status[sdk_internal_event] = status
+        
+    def _set_sdk_event_triggered(self, sdk_event):
+        if self._active_subscriptions.get(sdk_event) == None:
+            return
+        
+        if self._active_subscriptions.get(sdk_event).triggered == True:
+            return
+        
+        self._active_subscriptions[sdk_event] = self._active_subscriptions[sdk_event]._replace(triggered = True)
+        
+    def _get_event_handler(self, sdk_event):
+        if self._active_subscriptions.get(sdk_event) == None:
+            return None
+        
+        return self._active_subscriptions.get(sdk_event).handler
+
+    def _get_sdk_event_if_applicable(self, sdk_internal_event):
+        final_sdk_event = ValidSdkEvent(None, False)
+        self._update_internal_event_status(sdk_internal_event, True)
+         
+        events_to_fire = []
+        require_any_sdk_event = self._check_require_any(sdk_internal_event)
+        if require_any_sdk_event.valid:
+            if (not self._set_sdk_event_triggered(require_any_sdk_event.sdk_event) and 
+                    self._execution_limit(require_any_sdk_event.sdk_event) == 1)  or \
+                    self._execution_limit(require_any_sdk_event.sdk_event) == -1:
+                final_sdk_event = final_sdk_event._replace(sdk_event = require_any_sdk_event.sdk_event, 
+                                                           valid = self._check_prerequisites(require_any_sdk_event.sdk_event) and \
+                                                            self._check_suppressed_by(require_any_sdk_event.sdk_event))
+                                
+                if final_sdk_event.valid:
+                    events_to_fire.append(final_sdk_event.sdk_event)
+                    
+        [events_to_fire.append(sdk_event) for sdk_event in self._check_require_all()]   
+        
+        return events_to_fire
+    
+    def _check_require_all(self):
+        events = []
+        for require_name, require_value in self._manager_config.require_all.items():
+            final_status = True
+            for val in require_value:
+                final_status &= self._get_internal_event_status(val)
+            
+            if final_status and  \
+                self._check_prerequisites(require_name) and \
+                ((not self._event_already_triggered(require_name) and 
+                self._execution_limit(require_name) == 1)  or \
+                self._execution_limit(require_name) == -1) and \
+                len(require_value) > 0:
+                    
+                events.append(require_name)
+        
+        return events
+    
+    def _check_prerequisites(self, sdk_event):
+        for name, value in self._manager_config.prerequisites.items():
+            for val in value:
+                if name == sdk_event and not self._event_already_triggered(val):
+                    return False
+        
+        return True
+    
+    def _check_suppressed_by(self, sdk_event):
+        for name, value in self._manager_config.suppressed_by.items():
+            for val in value:
+                if name == sdk_event and self._event_already_triggered(val):
+                    return False
+        
+        return True
+
+    def _execution_limit(self, sdk_event):
+        limit = self._manager_config.execution_limits.get(sdk_event)
+        if limit == None:
+            return -1
+        
+        return limit
+    
+    def _check_require_any(self, sdk_internal_event):
+        valid_sdk_event = ValidSdkEvent(None, False)
+        for name, val in self._manager_config.require_any.items():
+            if sdk_internal_event in val:
+                valid_sdk_event = valid_sdk_event._replace(valid = True, sdk_event = name)
+                return valid_sdk_event
+        
+        return valid_sdk_event

tests/events/test_events_delivery.py

@@ -0,0 +1,27 @@
+"""EventsManager test module."""
+from splitio.models.events import SdkEvent, SdkInternalEvent
+from splitio.events.events_metadata import EventsMetadata
+from splitio.events.events_delivery import EventsDelivery
+from splitio.events.events_metadata import SdkEventType
+
+class EventsDeliveryTests(object):
+    """Tests for EventsManager."""
+
+    sdk_ready_flag = False
+    metadata = None
+    
+    def test_firing_events(self):
+        events_delivery = EventsDelivery()
+        
+        metadata = EventsMetadata(SdkEventType.FLAG_UPDATE, { "feature1" })
+        events_delivery.deliver(SdkEvent.SDK_READY, metadata, self._sdk_ready_callback)
+        assert self.sdk_ready_flag
+        self._verify_metadata(metadata)
+                    
+    def _sdk_ready_callback(self, metadata):
+        self.sdk_ready_flag = True
+        self.metadata = metadata
+
+    def _verify_metadata(self, metadata):
+        assert metadata.get_type() == self.metadata.get_type()
+        assert metadata.get_names() == self.metadata.get_names()

tests/events/test_events_manager.py

@@ -0,0 +1,100 @@
+"""EventsManager test module."""
+import pytest
+from splitio.models.events import SdkEvent, SdkInternalEvent
+from splitio.events.events_metadata import EventsMetadata
+from splitio.events.events_manager_config import EventsManagerConfig
+from splitio.events.events_delivery import EventsDelivery
+from splitio.events.events_manager import EventsManager
+from splitio.events.events_metadata import SdkEventType
+
+class EventsManagerTests(object):
+    """Tests for EventsManager."""
+
+    sdk_ready_flag = False
+    sdk_timed_out_flag = False
+    sdk_update_flag = False
+    metadata = None
+    
+    def test_firing_events(self):
+        events_manager = EventsManager(EventsManagerConfig(), EventsDelivery())
+        events_manager.register(SdkEvent.SDK_READY, self._sdk_ready_callback)
+        events_manager.register(SdkEvent.SDK_UPDATE, self._sdk_update_callback)
+        
+        metadata = EventsMetadata(SdkEventType.FLAG_UPDATE, { "feature1" })
+        events_manager.notify_internal_event(SdkInternalEvent.FLAGS_UPDATED, metadata)    
+        events_manager.notify_internal_event(SdkInternalEvent.FLAG_KILLED_NOTIFICATION, metadata)    
+        events_manager.notify_internal_event(SdkInternalEvent.RB_SEGMENTS_UPDATED, metadata)    
+        events_manager.notify_internal_event(SdkInternalEvent.SEGMENTS_UPDATED, metadata)    
+        assert not self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag
+        assert not self.sdk_update_flag
+        
+        self._reset_flags()
+        events_manager.notify_internal_event(SdkInternalEvent.SDK_TIMED_OUT, metadata)    
+        assert not self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag # not registered yet
+        assert not self.sdk_update_flag
+        
+        events_manager.register(SdkEvent.SDK_READY_TIMED_OUT, self._sdk_timeout_callback)
+        events_manager.notify_internal_event(SdkInternalEvent.SDK_TIMED_OUT, metadata)    
+        assert not self.sdk_ready_flag
+        assert self.sdk_timed_out_flag
+        assert not self.sdk_update_flag
+        self._verify_metadata(metadata)
+
+        self._reset_flags()
+        events_manager.notify_internal_event(SdkInternalEvent.SDK_READY, metadata)    
+        assert self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag
+        assert not self.sdk_update_flag
+        self._verify_metadata(metadata)
+
+        self._reset_flags()
+        events_manager.notify_internal_event(SdkInternalEvent.RB_SEGMENTS_UPDATED, metadata)    
+        assert not self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag
+        assert self.sdk_update_flag
+        self._verify_metadata(metadata)
+        
+        self._reset_flags()
+        events_manager.notify_internal_event(SdkInternalEvent.FLAG_KILLED_NOTIFICATION, metadata)    
+        assert not self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag
+        assert self.sdk_update_flag
+        self._verify_metadata(metadata)
+
+        self._reset_flags()
+        events_manager.notify_internal_event(SdkInternalEvent.FLAGS_UPDATED, metadata)    
+        assert not self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag
+        assert self.sdk_update_flag
+        self._verify_metadata(metadata)
+
+        self._reset_flags()
+        events_manager.notify_internal_event(SdkInternalEvent.SEGMENTS_UPDATED, metadata)    
+        assert not self.sdk_ready_flag
+        assert not self.sdk_timed_out_flag
+        assert self.sdk_update_flag
+        self._verify_metadata(metadata)
+    
+    def _reset_flags(self):
+        self.sdk_ready_flag = False
+        self.sdk_timed_out_flag = False
+        self.sdk_update_flag = False
+        self.metadata = None
+        
+    def _sdk_ready_callback(self, metadata):
+        self.sdk_ready_flag = True
+        self.metadata = metadata
+
+    def _sdk_update_callback(self, metadata):
+        self.sdk_update_flag = True
+        self.metadata = metadata
+
+    def _sdk_timeout_callback(self, metadata):
+        self.sdk_timed_out_flag = True
+        self.metadata = metadata
+
+    def _verify_metadata(self, metadata):
+        assert metadata.get_type() == self.metadata.get_type()
+        assert metadata.get_names() == self.metadata.get_names()