Merge pull request #207 from splitio/task/fixes

added shutdown method

Author: mmelograno

splitio/client/factory.py

@@ -213,6 +213,8 @@ def _wait_for_tasks_to_stop():
                     wait_thread.start()
                 else:
                     self._sync_manager.stop(False)
+            elif destroyed_event is not None:
+                destroyed_event.set()
         finally:
             self._status = Status.DESTROYED
             with _INSTANTIATED_FACTORIES_LOCK:

splitio/sync/manager.py

@@ -73,8 +73,7 @@ def stop(self, blocking):
             self._push_status_handler_active = False
             self._queue.put(self._CENTINEL_EVENT)
             self._push.stop()
-        self._synchronizer.stop_periodic_fetching(True)
-        self._synchronizer.stop_periodic_data_recording(blocking)
+        self._synchronizer.shutdown(blocking)
 
     def _streaming_feedback_handler(self):
         """

splitio/sync/synchronizer.py

@@ -170,13 +170,8 @@ def start_periodic_fetching(self):
         pass
 
     @abc.abstractmethod
-    def stop_periodic_fetching(self, shutdown=False):
-        """
-        Stop fetchers for splits and segments.
-
-        :param shutdown: flag to indicates if should pause or stop tasks
-        :type shutdown: bool
-        """
+    def stop_periodic_fetching(self):
+        """Stop fetchers for splits and segments."""
         pass
 
     @abc.abstractmethod
@@ -203,6 +198,16 @@ def kill_split(self, split_name, default_treatment, change_number):
         """
         pass
 
+    @abc.abstractmethod
+    def shutdown(self, blocking):
+        """
+        Stop tasks
+
+        :param blocking:flag to wait until tasks are stopped
+        :type blocking: bool
+        """
+        pass
+
 
 class Synchronizer(BaseSynchronizer):
     """Synchronizer."""
@@ -256,23 +261,28 @@ def sync_all(self):
             _LOGGER.error('Failed syncing splits')
             raise_from(APIException('Failed to sync splits'), exc)
 
+    def shutdown(self, blocking):
+        """
+        Stop tasks
+
+        :param blocking:flag to wait until tasks are stopped
+        :type blocking: bool
+        """
+        _LOGGER.debug('Shutting down tasks.')
+        self._split_synchronizers.segment_sync.shutdown()
+        self.stop_periodic_fetching()
+        self.stop_periodic_data_recording(blocking)
+
     def start_periodic_fetching(self):
         """Start fetchers for splits and segments."""
         _LOGGER.debug('Starting periodic data fetching')
         self._split_tasks.split_task.start()
         self._split_tasks.segment_task.start()
 
-    def stop_periodic_fetching(self, shutdown=False):
-        """
-        Stop fetchers for splits and segments.
-
-        :param shutdown: flag to indicates if should pause or stop tasks
-        :type shutdown: bool
-        """
+    def stop_periodic_fetching(self):
+        """Stop fetchers for splits and segments."""
         _LOGGER.debug('Stopping periodic fetching')
         self._split_tasks.split_task.stop()
-        if shutdown:  # stops task and worker pool
-            self._split_synchronizers.segment_sync.shutdown()
         self._split_tasks.segment_task.stop()
 
     def start_periodic_data_recording(self):
@@ -352,7 +362,7 @@ def start_periodic_fetching(self):
         _LOGGER.debug('Starting periodic data fetching')
         self._split_tasks.split_task.start()
 
-    def stop_periodic_fetching(self, shutdown=False):
+    def stop_periodic_fetching(self):
         """Stop fetchers for splits and segments."""
         _LOGGER.debug('Stopping periodic fetching')
         self._split_tasks.split_task.stop()

tests/client/test_factory.py

@@ -330,6 +330,35 @@ def _split_synchronizer(self, ready_flag, some, auth_api, streaming_enabled, sse
         assert len(imp_count_async_task_mock.stop.mock_calls) == 1
         assert factory.destroyed is True
 
+    def test_destroy_with_event_redis(self, mocker):
+        def _make_factory_with_apikey(apikey, *_, **__):
+            return SplitFactory(apikey, {}, True, mocker.Mock(spec=ImpressionsManager), None)
+
+        factory_module_logger = mocker.Mock()
+        build_redis = mocker.Mock()
+        build_redis.side_effect = _make_factory_with_apikey
+        mocker.patch('splitio.client.factory._LOGGER', new=factory_module_logger)
+        mocker.patch('splitio.client.factory._build_redis_factory', new=build_redis)
+
+        config = {
+            'redisDb': 0,
+            'redisHost': 'localhost',
+            'redisPosrt': 6379,
+        }
+
+        factory = get_factory("none", config=config)
+        event = threading.Event()
+        factory.destroy(event)
+        event.wait()
+        assert factory.destroyed
+        assert len(build_redis.mock_calls) == 1
+
+        factory = get_factory("none", config=config)
+        factory.destroy(None)
+        time.sleep(0.1)
+        assert factory.destroyed
+        assert len(build_redis.mock_calls) == 2
+
     def test_multiple_factories(self, mocker):
         """Test multiple factories instantiation and tracking."""
         sdk_ready_flag = threading.Event()

tests/sync/test_synchronizer.py

@@ -124,17 +124,11 @@ def test_stop_periodic_fetching(self, mocker):
         split_tasks = SplitTasks(split_task, segment_task, mocker.Mock(), mocker.Mock(),
                                  mocker.Mock(), mocker.Mock())
         synchronizer = Synchronizer(split_synchronizers, split_tasks)
-        synchronizer.stop_periodic_fetching(True)
+        synchronizer.stop_periodic_fetching()
 
         assert len(split_task.stop.mock_calls) == 1
         assert len(segment_task.stop.mock_calls) == 1
-        assert len(segment_sync.shutdown.mock_calls) == 1
-
-        synchronizer.stop_periodic_fetching(False)
-
-        assert len(split_task.stop.mock_calls) == 2
-        assert len(segment_task.stop.mock_calls) == 2
-        assert len(segment_sync.shutdown.mock_calls) == 1  # not called here
+        assert len(segment_sync.shutdown.mock_calls) == 0
 
     def test_start_periodic_data_recording(self, mocker):
         impression_task = mocker.Mock(spec=ImpressionsSyncTask)
@@ -177,3 +171,42 @@ def stop_mock_2():
         assert len(impression_count_task.stop.mock_calls) == 1
         assert len(event_task.stop.mock_calls) == 1
         assert len(telemetry_task.stop.mock_calls) == 1
+
+    def test_shutdown(self, mocker):
+
+        def stop_mock(event):
+            event.set()
+            return
+
+        def stop_mock_2():
+            return
+
+        split_task = mocker.Mock(spec=SplitSynchronizationTask)
+        split_task.stop.side_effect = stop_mock_2
+        segment_task = mocker.Mock(spec=SegmentSynchronizationTask)
+        segment_task.stop.side_effect = stop_mock_2
+        impression_task = mocker.Mock(spec=ImpressionsSyncTask)
+        impression_task.stop.side_effect = stop_mock
+        impression_count_task = mocker.Mock(spec=ImpressionsCountSyncTask)
+        impression_count_task.stop.side_effect = stop_mock
+        event_task = mocker.Mock(spec=EventsSyncTask)
+        event_task.stop.side_effect = stop_mock
+        telemetry_task = mocker.Mock(spec=TelemetrySynchronizationTask)
+        telemetry_task.stop.side_effect = stop_mock_2
+
+        segment_sync = mocker.Mock(spec=SegmentSynchronizer)
+
+        split_synchronizers = SplitSynchronizers(mocker.Mock(), segment_sync, mocker.Mock(),
+                                                 mocker.Mock(), mocker.Mock(), mocker.Mock())
+        split_tasks = SplitTasks(split_task, segment_task, impression_task, event_task,
+                                 telemetry_task, impression_count_task)
+        synchronizer = Synchronizer(split_synchronizers, split_tasks)
+        synchronizer.shutdown(True)
+
+        assert len(split_task.stop.mock_calls) == 1
+        assert len(segment_task.stop.mock_calls) == 1
+        assert len(segment_sync.shutdown.mock_calls) == 1
+        assert len(impression_task.stop.mock_calls) == 1
+        assert len(impression_count_task.stop.mock_calls) == 1
+        assert len(event_task.stop.mock_calls) == 1
+        assert len(telemetry_task.stop.mock_calls) == 1