Merge pull request #210 from splitio/tests/push_integration

Tests/push integration

Author: mredolatti

splitio/client/config.py

@@ -16,8 +16,8 @@
     'splitSdkMachineName': None,
     'splitSdkMachineIp': None,
     'streamingEnabled': True,
-    'featuresRefreshRate': 5,
-    'segmentsRefreshRate': 60,
+    'featuresRefreshRate': 30,
+    'segmentsRefreshRate': 30,
     'metricsRefreshRate': 60,
     'impressionsRefreshRate': 5 * 60,
     'impressionsBulkSize': 5000,

splitio/push/manager.py

@@ -51,84 +51,46 @@ def __init__(self, auth_api, synchronizer, feedback_loop, sse_url=None):
             MessageType.OCCUPANCY: self._handle_occupancy
         }
 
-        self._sse_client = SplitSSEClient(self._event_handler) if sse_url is None \
-            else SplitSSEClient(self._event_handler, sse_url)
+        kwargs = {} if sse_url is None else {'base_url': sse_url}
+        self._sse_client = SplitSSEClient(self._event_handler, self._handle_connection_ready,
+                                          self._handle_connection_end, **kwargs)
         self._running = False
         self._next_refresh = Timer(0, lambda: 0)
 
-    def _handle_message(self, event):
-        """
-        Handle incoming update message.
-
-        :param event: Incoming Update message
-        :type event: splitio.push.sse.parser.Update
-        """
-        try:
-            handle = self._message_handlers[event.message_type]
-        except KeyError:
-            _LOGGER.error('no handler for message of type %s', event.message_type)
-            _LOGGER.debug(str(event), exc_info=True)
-            return
-
-        handle(event)
-
-    def _handle_update(self, event):
-        """
-        Handle incoming update message.
-
-        :param event: Incoming Update message
-        :type event: splitio.push.sse.parser.Update
-        """
-        _LOGGER.debug('handling update event: %s', str(event))
-        self._processor.handle(event)
-
-    def _handle_control(self, event):
-        """
-        Handle incoming control message.
-
-        :param event: Incoming control message.
-        :type event: splitio.push.sse.parser.ControlMessage
+    def update_workers_status(self, enabled):
         """
-        _LOGGER.debug('handling control event: %s', str(event))
-        feedback = self._status_tracker.handle_control_message(event)
-        if feedback is not None:
-            self._feedback_loop.put(feedback)
+        Enable/Disable push update workers.
 
-    def _handle_occupancy(self, event):
+        :param enabled: if True, enable workers. If False, disable them.
+        :type enabled: bool
         """
-        Handle incoming notification message.
+        self._processor.update_workers_status(enabled)
 
-        :param event: Incoming occupancy message.
-        :type event: splitio.push.sse.parser.Occupancy
-        """
-        _LOGGER.debug('handling occupancy event: %s', str(event))
-        feedback = self._status_tracker.handle_occupancy(event)
-        if feedback is not None:
-            self._feedback_loop.put(feedback)
 
-    def _handle_connection_end(self, shutdown_requested):
-        """
-        Handle a connection ending.
+    def start(self):
+        """Start a new connection if not already running."""
+        if self._running:
+            _LOGGER.warning('Push manager already has a connection running. Ignoring')
+            return
 
-        If the connection shutdown was not requested, trigger a restart.
+        self._trigger_connection_flow()
 
-        :param shutdown_requested: whether the shutdown was requested or unexpected.
-        :type shutdown_requested: True
+    def stop(self, blocking=False):
         """
-        if not shutdown_requested:
-            self._feedback_loop.put(Status.PUSH_RETRYABLE_ERROR)
+        Stop the current ongoing connection.
 
-    def _handle_error(self, event):
+        :param blocking: whether to wait for the connection to be successfully closed or not
+        :type blocking: bool
         """
-        Handle incoming error message.
+        if not self._running:
+            _LOGGER.warning('Push manager does not have an open SSE connection. Ignoring')
+            return
 
-        :param event: Incoming ably error
-        :type event: splitio.push.sse.parser.AblyError
-        """
-        _LOGGER.debug('handling ably error event: %s', str(event))
-        feedback = self._status_tracker.handle_ably_error(event)
-        if feedback is not None:
-            self._feedback_loop.put(feedback)
+        self._running = False
+        self._processor.update_workers_status(False)
+        self._status_tracker.notify_sse_shutdown_expected()
+        self._next_refresh.cancel()
+        self._sse_client.stop(blocking)
 
     def _event_handler(self, event):
         """
@@ -178,14 +140,12 @@ def _trigger_connection_flow(self):
             self._feedback_loop.put(Status.PUSH_NONRETRYABLE_ERROR)
             return
 
+        _LOGGER.debug("auth token fetched. connecting to streaming.")
         self._status_tracker.reset()
-        if  self._sse_client.start(token):
+        if self._sse_client.start(token):
+            _LOGGER.debug("connected to streaming, scheduling next refresh")
             self._setup_next_token_refresh(token)
             self._running = True
-            self._feedback_loop.put(Status.PUSH_SUBSYSTEM_UP)
-            return
-
-        self._feedback_loop.put(Status.PUSH_RETRYABLE_ERROR)
 
     def _setup_next_token_refresh(self, token):
         """
@@ -201,36 +161,79 @@ def _setup_next_token_refresh(self, token):
         self._next_refresh.setName('TokenRefresh')
         self._next_refresh.start()
 
-    def update_workers_status(self, enabled):
+    def _handle_message(self, event):
         """
-        Enable/Disable push update workers.
+        Handle incoming update message.
 
-        :param enabled: if True, enable workers. If False, disable them.
-        :type enabled: bool
+        :param event: Incoming Update message
+        :type event: splitio.push.sse.parser.Update
         """
-        self._processor.update_workers_status(enabled)
+        try:
+            handle = self._message_handlers[event.message_type]
+        except KeyError:
+            _LOGGER.error('no handler for message of type %s', event.message_type)
+            _LOGGER.debug(str(event), exc_info=True)
+            return
 
+        handle(event)
 
-    def start(self):
-        """Start a new connection if not already running."""
-        if self._running:
-            _LOGGER.warning('Push manager already has a connection running. Ignoring')
-            return
+    def _handle_update(self, event):
+        """
+        Handle incoming update message.
 
-        self._trigger_connection_flow()
+        :param event: Incoming Update message
+        :type event: splitio.push.sse.parser.Update
+        """
+        _LOGGER.debug('handling update event: %s', str(event))
+        self._processor.handle(event)
 
-    def stop(self, blocking=False):
+    def _handle_control(self, event):
         """
-        Stop the current ongoing connection.
+        Handle incoming control message.
 
-        :param blocking: whether to wait for the connection to be successfully closed or not
-        :type blocking: bool
+        :param event: Incoming control message.
+        :type event: splitio.push.sse.parser.ControlMessage
         """
-        if not self._running:
-            _LOGGER.warning('Push manager does not have an open SSE connection. Ignoring')
-            return
+        _LOGGER.debug('handling control event: %s', str(event))
+        feedback = self._status_tracker.handle_control_message(event)
+        if feedback is not None:
+            self._feedback_loop.put(feedback)
 
-        self._processor.update_workers_status(False)
-        self._status_tracker.notify_sse_shutdown_expected()
-        self._next_refresh.cancel()
-        self._sse_client.stop(blocking)
+    def _handle_occupancy(self, event):
+        """
+        Handle incoming notification message.
+
+        :param event: Incoming occupancy message.
+        :type event: splitio.push.sse.parser.Occupancy
+        """
+        _LOGGER.debug('handling occupancy event: %s', str(event))
+        feedback = self._status_tracker.handle_occupancy(event)
+        if feedback is not None:
+            self._feedback_loop.put(feedback)
+
+    def _handle_error(self, event):
+        """
+        Handle incoming error message.
+
+        :param event: Incoming ably error
+        :type event: splitio.push.sse.parser.AblyError
+        """
+        _LOGGER.debug('handling ably error event: %s', str(event))
+        feedback = self._status_tracker.handle_ably_error(event)
+        if feedback is not None:
+            self._feedback_loop.put(feedback)
+
+    def _handle_connection_ready(self):
+        """Handle a successful connection to SSE."""
+        self._feedback_loop.put(Status.PUSH_SUBSYSTEM_UP)
+        _LOGGER.info('sse initial event received. enabling')
+
+    def _handle_connection_end(self):
+        """
+        Handle a connection ending.
+
+        If the connection shutdown was not requested, trigger a restart.
+        """
+        feedback = self._status_tracker.handle_disconnect()
+        if feedback is not None:
+            self._feedback_loop.put(feedback)

splitio/push/parser.py

@@ -1,13 +1,13 @@
 """SSE Notification definitions."""
 import abc
 import json
-import time
 from enum import Enum
 
 from future.utils import raise_from
 from six import add_metaclass
 
 from splitio.util.decorators import abstract_property
+from splitio.util import utctime_ms
 from splitio.push.sse import SSE_EVENT_ERROR, SSE_EVENT_MESSAGE
 
 
@@ -37,9 +37,9 @@ class UpdateType(Enum):
 class ControlType(Enum):
     """Control type enumeration."""
 
-    STREAMING_ENABLED = 0
-    STREAMING_PAUSED = 1
-    STREAMING_DISABLED = 2
+    STREAMING_ENABLED = 'STREAMING_ENABLED'
+    STREAMING_PAUSED = 'STREAMING_PAUSED'
+    STREAMING_DISABLED = 'STREAMING_DISABLED'
 
 
 TAG_OCCUPANCY = '[meta]occupancy'
@@ -89,7 +89,7 @@ def __init__(self, code, status_code, message, href):
         self._status_code = status_code
         self._message = message
         self._href = href
-        self._timestamp = int(time.time() * 1000)  # TODO: replace with UTC function after merge
+        self._timestamp = utctime_ms()
 
     @property
     def event_type(self):  #pylint:disable=no-self-use

splitio/push/splitsse.py

@@ -10,7 +10,7 @@
 _LOGGER = logging.getLogger(__name__)
 
 
-class SplitSSEClient(object):
+class SplitSSEClient(object):  # pylint: disable=too-many-instance-attributes
     """Split streaming endpoint SSE client."""
 
     KEEPALIVE_TIMEOUT = 70
@@ -21,18 +21,27 @@ class _Status(Enum):
         ERRORED = 2
         CONNECTED = 3
 
-    def __init__(self, callback, base_url='https://streaming.split.io'):
+    def __init__(self, event_callback, first_event_callback=None,
+                 connection_closed_callback=None, base_url='https://streaming.split.io'):
         """
         Construct a split sse client.
 
         :param callback: fuction to call when an event is received.
         :type callback: callable
 
+        :param first_event_callback: function to call when the first event is received.
+        :type first_event_callback: callable
+
+        :param connection_closed_callback: funciton to call when the connection ends.
+        :type connection_closed_callback: callable
+
         :param base_url: scheme + :// + host
         :type base_url: str
         """
         self._client = SSEClient(self._raw_event_handler)
-        self._callback = callback
+        self._callback = event_callback
+        self._on_connected = first_event_callback
+        self._on_disconnected = connection_closed_callback
         self._base_url = base_url
         self._status = SplitSSEClient._Status.IDLE
         self._sse_first_event = None
@@ -49,6 +58,8 @@ def _raw_event_handler(self, event):
             self._status = SplitSSEClient._Status.CONNECTED if event.event != SSE_EVENT_ERROR \
                 else SplitSSEClient._Status.ERRORED
             self._sse_first_event.set()
+            if self._on_connected is not None:
+                self._on_connected()
 
         if event.data is not None:
             self._callback(event)
@@ -111,6 +122,7 @@ def connect(url):
             finally:
                 self._status = SplitSSEClient._Status.IDLE
                 self._sse_connection_closed.set()
+                self._on_disconnected()
 
         url = self._build_url(token)
         task = threading.Thread(target=connect, name='SSEConnection', args=(url,))

splitio/push/splitworker.py

@@ -41,7 +41,7 @@ def _run(self):
             _LOGGER.debug('Processing split_update %d', event.change_number)
             try:
                 self._handler(event.change_number)
-            except Exception:
+            except Exception:  # pylint: disable=broad-except
                 _LOGGER.error('Exception raised in split synchronization')
                 _LOGGER.debug('Exception information: ', exc_info=True)
 

splitio/push/status_tracker.py

@@ -1,5 +1,4 @@
 """NotificationManagerKeeper implementation."""
-from collections import defaultdict
 from enum import Enum
 import logging
 import six
@@ -18,7 +17,7 @@ class Status(Enum):
     PUSH_NONRETRYABLE_ERROR = 3
 
 
-class LastEventTimestamps(object):
+class LastEventTimestamps(object):  # pylint:disable=too-few-public-methods
     """Simple class to keep track of the last time multiple events occurred."""
 
     def __init__(self):
@@ -90,7 +89,7 @@ def handle_control_message(self, event):
         :type event: splitio.push.parser.ControlMessage
         """
         # we don't care about control messages if a disconnection is expected
-        if self._shutdown_expected:  
+        if self._shutdown_expected:
             return None
 
         if self._timestamps.control > event.timestamp:
@@ -111,10 +110,10 @@ def handle_ably_error(self, event):
         :returns: A new status if required. None otherwise
         :rtype: Optional[Status]
         """
-        if self._shutdown_expected:  # we don't care about occupancy if a disconnection is expected
+        if self._shutdown_expected:  # we don't care about an incoming error if a shutdown is expected
             return None
 
-        _LOGGER.debug('handling update event: %s', str(event))
+        _LOGGER.debug('handling ably error event: %s', str(event))
         if event.should_be_ignored():
             _LOGGER.debug('ignoring sse error message: %s', event)
             return None
@@ -161,7 +160,7 @@ def _update_status(self):
 
         return None
 
-    def _handle_disconnect(self):
+    def handle_disconnect(self):
         """
         Handle non-requested SSE disconnection.