Merge pull request #206 from splitio/task/fixes

Task/fixes

Author: mmelograno

splitio/client/factory.py

@@ -300,7 +300,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/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

@@ -272,9 +272,8 @@ def stop_periodic_fetching(self, shutdown=False):
         _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_synchronizers.segment_sync.shutdown()
+        self._split_tasks.segment_task.stop()
 
     def start_periodic_data_recording(self):
         """Start recorders."""

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)