Merge branch 'feature/streaming' of github.com:splitio/python-client into tests/sdkmockserver

Author: mredolatti

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:
@@ -300,7 +302,6 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         ),
         SegmentSynchronizationTask(
             synchronizers.segment_sync.synchronize_segments,
-            synchronizers.segment_sync.worker_pool,
             cfg['segmentsRefreshRate'],
         ),
         ImpressionsSyncTask(

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/segment.py

@@ -30,16 +30,12 @@ def __init__(self, segment_api, split_storage, segment_storage):
         self._worker_pool = workerpool.WorkerPool(10, self.synchronize_segment)
         self._worker_pool.start()
 
-    @property
-    def worker_pool(self):
+    def shutdown(self):
         """
-        Return worker_pool
-
-        :return: workerpool
-        :rtype: splitio.tasks.util.WorkerPool
+        Shutdown worker_pool
 
         """
-        return self._worker_pool
+        self._worker_pool.stop()
 
     def synchronize_segment(self, segment_name, till=None):
         """
@@ -78,8 +74,7 @@ def synchronize_segment(self, segment_name, till=None):
                     segment_changes['till']
                 )
 
-            if segment_changes['till'] == segment_changes['since'] \
-               or (till is not None and segment_changes['till'] >= till):
+            if segment_changes['till'] == segment_changes['since']:
                 return
 
     def synchronize_segments(self):

splitio/sync/split.py

@@ -60,7 +60,7 @@ def synchronize_splits(self, till=None):
 
             self._split_storage.set_change_number(split_changes['till'])
             if split_changes['till'] == split_changes['since'] \
-               or (till is not None and split_changes['till'] >= till):
+               and (till is None or split_changes['till'] >= till):
                 return
 
     def kill_split(self, split_name, default_treatment, change_number):

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,25 +261,29 @@ 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_tasks.segment_task.stop()
-        else:  # pauses task not worker pool
-            self._split_tasks.segment_task.pause()
+        self._split_tasks.segment_task.stop()
 
     def start_periodic_data_recording(self):
         """Start recorders."""
@@ -353,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()

splitio/tasks/segment_sync.py

@@ -3,7 +3,7 @@
 import logging
 from splitio.api import APIException
 from splitio.tasks import BaseSynchronizationTask
-from splitio.tasks.util import asynctask, workerpool
+from splitio.tasks.util import asynctask
 
 
 _LOGGER = logging.getLogger(__name__)
@@ -12,33 +12,23 @@
 class SegmentSynchronizationTask(BaseSynchronizationTask):
     """Segment Syncrhonization class."""
 
-    def __init__(self, synchronize_segments, worker_pool, period):
+    def __init__(self, synchronize_segments, period):
         """
         Clas constructor.
 
         :param synchronize_segments: handler for syncing segments
         :type synchronize_segments: func
 
-        :param worker_pool: worker created by sync to be able to stop worker
-        :type worker_pool:  splitio.tasks.util.WorkerPool
-
         """
-        self._worker_pool = worker_pool
         self._task = asynctask.AsyncTask(synchronize_segments, period, on_init=None)
 
     def start(self):
         """Start segment synchronization."""
         self._task.start()
 
-    def pause(self):
-        """Pause segment synchronization."""
-        self._task.stop()
-
     def stop(self, event=None):
         """Stop segment synchronization."""
-        self._task.stop()
-        if self._worker_pool is not None:
-            self._worker_pool.stop(event)
+        self._task.stop(event)
 
     def is_running(self):
         """

splitio/tasks/split_sync.py

@@ -20,7 +20,7 @@ def __init__(self, synchronize_splits, period):
         :type period: int
         """
         self._period = period
-        self._task = AsyncTask(synchronize_splits, period, on_init=synchronize_splits)
+        self._task = AsyncTask(synchronize_splits, period, on_init=None)
 
     def start(self):
         """Start the task."""

tests/client/test_factory.py

@@ -8,7 +8,7 @@
 from splitio.client.config import DEFAULT_CONFIG
 from splitio.storage import redis, inmemmory, uwsgi
 from splitio.tasks import events_sync, impressions_sync, split_sync, segment_sync, telemetry_sync
-from splitio.tasks.util import asynctask, workerpool
+from splitio.tasks.util import asynctask
 from splitio.api.splits import SplitsAPI
 from splitio.api.segments import SegmentsAPI
 from splitio.api.impressions import ImpressionsAPI
@@ -161,9 +161,8 @@ def _split_task_init_mock(self, synchronize_splits, period):
         segment_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
         segment_async_task_mock.stop.side_effect = stop_mock
 
-        def _segment_task_init_mock(self, synchronize_segments, worker_pool, period):
+        def _segment_task_init_mock(self, synchronize_segments, period):
             self._task = segment_async_task_mock
-            self._worker_pool = mocker.Mock()
             self._period = period
         mocker.patch('splitio.client.factory.SegmentSynchronizationTask.__init__',
                      new=_segment_task_init_mock)
@@ -256,9 +255,8 @@ def _split_task_init_mock(self, synchronize_splits, period):
         segment_async_task_mock = mocker.Mock(spec=asynctask.AsyncTask)
         segment_async_task_mock.stop.side_effect = stop_mock_2
 
-        def _segment_task_init_mock(self, synchronize_segments, worker_pool, period):
+        def _segment_task_init_mock(self, synchronize_segments, period):
             self._task = segment_async_task_mock
-            self._worker_pool = mocker.Mock()
             self._period = period
         mocker.patch('splitio.client.factory.SegmentSynchronizationTask.__init__',
                      new=_segment_task_init_mock)
@@ -332,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()