more integration tests and UT fixing

Author: mredolatti

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):
+    def update_workers_status(self, enabled):
         """
-        Handle incoming control message.
+        Enable/Disable push update workers.
 
-        :param event: Incoming control message.
-        :type event: splitio.push.sse.parser.ControlMessage
+        :param enabled: if True, enable workers. If False, disable them.
+        :type enabled: bool
         """
-        _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(enabled)
 
-    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_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):
         """
@@ -179,13 +141,11 @@ def _trigger_connection_flow(self):
             return
 
         self._status_tracker.reset()
-        if  self._sse_client.start(token):
+        res = self._sse_client.start(token)
+        if res:
+            print(res)
             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,81 @@ def _setup_next_token_refresh(self, token):
         self._next_refresh.setName('TokenRefresh')
         self._next_refresh.start()
 
-    def update_workers_status(self, enabled):
+    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, shutdown_requested):
         """
-        Enable/Disable push update workers.
+        Handle a connection ending.
 
-        :param enabled: if True, enable workers. If False, disable them.
-        :type enabled: bool
+        If the connection shutdown was not requested, trigger a restart.
+
+        :param shutdown_requested: whether the shutdown was requested or unexpected.
+        :type shutdown_requested: True
         """
-        self._processor.update_workers_status(enabled)
+        if not shutdown_requested:
+            self._feedback_loop.put(Status.PUSH_RETRYABLE_ERROR)
 
+    def _handle_message(self, event):
+        """
+        Handle incoming update message.
 
-    def start(self):
-        """Start a new connection if not already running."""
-        if self._running:
-            _LOGGER.warning('Push manager already has a connection running. Ignoring')
+        :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
 
-        self._trigger_connection_flow()
+        handle(event)
 
-    def stop(self, blocking=False):
+    def _handle_update(self, event):
         """
-        Stop the current ongoing connection.
+        Handle incoming update message.
 
-        :param blocking: whether to wait for the connection to be successfully closed or not
-        :type blocking: bool
+        :param event: Incoming Update message
+        :type event: splitio.push.sse.parser.Update
         """
-        if not self._running:
-            _LOGGER.warning('Push manager does not have an open SSE connection. Ignoring')
-            return
+        _LOGGER.debug('handling update event: %s', str(event))
+        self._processor.handle(event)
 
-        self._processor.update_workers_status(False)
-        self._status_tracker.notify_sse_shutdown_expected()
-        self._next_refresh.cancel()
-        self._sse_client.stop(blocking)
+    def _handle_control(self, event):
+        """
+        Handle incoming control message.
+
+        :param event: Incoming control message.
+        :type event: splitio.push.sse.parser.ControlMessage
+        """
+        _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)
+
+    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)

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)
@@ -106,11 +117,13 @@ def start(self, token):
 
         def connect(url):
             """Connect to sse in a blocking manner."""
+            shutdown_requested = False
             try:
-                self._client.start(url, timeout=self.KEEPALIVE_TIMEOUT)
+                shutdown_requested = self._client.start(url, timeout=self.KEEPALIVE_TIMEOUT)
             finally:
                 self._status = SplitSSEClient._Status.IDLE
                 self._sse_connection_closed.set()
+                self._on_disconnected(shutdown_requested)
 
         url = self._build_url(token)
         task = threading.Thread(target=connect, name='SSEConnection', args=(url,))

splitio/sync/manager.py

@@ -88,23 +88,24 @@ def _streaming_feedback_handler(self):
                 continue
 
             if status == Status.PUSH_SUBSYSTEM_UP:
-                _LOGGER.info('streaming up and running. disabling periodic fetching.')
                 self._synchronizer.stop_periodic_fetching()
                 self._synchronizer.sync_all()
                 self._push.update_workers_status(True)
                 self._backoff.reset()
+                _LOGGER.info('streaming up and running. disabling periodic fetching.')
             elif status == Status.PUSH_SUBSYSTEM_DOWN:
-                _LOGGER.info('streaming temporarily down. starting periodic fetching')
                 self._push.update_workers_status(False)
+                self._synchronizer.sync_all()
                 self._synchronizer.start_periodic_fetching()
+                _LOGGER.info('streaming temporarily down. starting periodic fetching')
             elif status == Status.PUSH_RETRYABLE_ERROR:
-                _LOGGER.info('error in streaming. restarting flow')
                 self._synchronizer.start_periodic_fetching()
                 self._push.stop(True)
                 time.sleep(self._backoff.get())
                 self._push.start()
+                _LOGGER.info('error in streaming. restarting flow')
             elif status == Status.PUSH_NONRETRYABLE_ERROR:
-                _LOGGER.info('non-recoverable error in streaming. switching to polling.')
                 self._synchronizer.start_periodic_fetching()
                 self._push.stop(False)
+                _LOGGER.info('non-recoverable error in streaming. switching to polling.')
                 return

splitio/tasks/util/asynctask.py

@@ -89,10 +89,10 @@ def _execution_wrapper(self):
                 try:
                     msg = self._messages.get(True, self._period)
                     if msg == __TASK_STOP__:
-                        _LOGGER.info("Stop signal received. finishing task execution")
+                        _LOGGER.debug("Stop signal received. finishing task execution")
                         break
                     elif msg == __TASK_FORCE_RUN__:
-                        _LOGGER.info("Force execution signal received. Running now")
+                        _LOGGER.debug("Force execution signal received. Running now")
                         if not _safe_run(self._main):
                             _LOGGER.error(
                                 "An error occurred when executing the task. "