fix error propagation

Author: mredolatti

splitio/push/manager.py

@@ -92,8 +92,7 @@ def _handle_control(self, event):
         _LOGGER.debug('handling control event: %s', str(event))
         feedback = self._status_tracker.handle_control_message(event)
         if feedback is not None:
-            # Send this event back to sync manager
-            pass
+            self._feedback_loop.put(feedback)
 
     def _handle_occupancy(self, event):
         """
@@ -105,8 +104,19 @@ def _handle_occupancy(self, event):
         _LOGGER.debug('handling occupancy event: %s', str(event))
         feedback = self._status_tracker.handle_occupancy(event)
         if feedback is not None:
-            # Send this event back to sync manager
-            pass
+            self._feedback_loop.put(feedback)
+
+    def _handle_connection_end(self, shutdown_requested):
+        """
+        Handle a connection ending.
+
+        If the connection shutdown was not requested, trigger a restart.
+
+        :param shutdown_requested: whether the shutdown was requested or unexpected.
+        :type shutdown_requested: True
+        """
+        if not shutdown_requested:
+            self._feedback_loop.put(Status.PUSH_RETRYABLE_ERROR)
 
     def _handle_error(self, event):
         """
@@ -118,8 +128,7 @@ def _handle_error(self, event):
         _LOGGER.debug('handling ably error event: %s', str(event))
         feedback = self._status_tracker.handle_ably_error(event)
         if feedback is not None:
-            # Send this event back to sync manager
-            pass
+            self._feedback_loop.put(feedback)
 
     def _event_handler(self, event):
         """
@@ -188,7 +197,7 @@ def _setup_next_token_refresh(self, token):
         if self._next_refresh is not None:
             self._next_refresh.cancel()
         self._next_refresh = Timer((token.exp - token.iat) - _TOKEN_REFRESH_GRACE_PERIOD,
-                                   self._token_refresh)
+                                   self._token_refresh, name='TokenRefresh')
         self._next_refresh.start()
 
     def update_workers_status(self, enabled):

tests/push/test_manager.py

@@ -32,7 +32,7 @@ def test_connection_success(self, mocker):
         assert timer_mock.mock_calls == [
             mocker.call(0, Any()),
             mocker.call().cancel(),
-            mocker.call(1000000 - _TOKEN_REFRESH_GRACE_PERIOD, manager._token_refresh),
+            mocker.call(1000000 - _TOKEN_REFRESH_GRACE_PERIOD, manager._token_refresh, name='TokenRefresh'),
             mocker.call().start()
         ]
 

tests/push/test_splitsse.py

@@ -72,8 +72,9 @@ def handler(event):
         server.publish({'event': 'error'})  # send an error event early to unblock start
         assert not client.start(token)
         client.stop(True)
-        with pytest.raises(Exception):
-            client.stop()
+
+        # should do nothing
+        client.stop()
 
         request = request_queue.get(1)
         assert request.path == '/event-stream?v=1.1&accessToken=some&channels=chan1,[?occupancy=metrics.publishers]chan2'