Merge pull request #384 from splitio/async-segment-worker

Updated SegmentWorker

Author: chillaq

splitio/push/processor.py

@@ -3,8 +3,8 @@
 from queue import Queue
 
 from splitio.push.parser import UpdateType
-from splitio.push.splitworker import SplitWorker
-from splitio.push.segmentworker import SegmentWorker
+from splitio.push.workers import SplitWorker
+from splitio.push.workers import SegmentWorker
 
 
 class MessageProcessor(object):

splitio/push/segmentworker.py

@@ -1,14 +1,15 @@
-"""Feature Flag changes processing worker."""
+"""Segment changes processing worker."""
 import logging
 import threading
 import abc
 
 from splitio.optional.loaders import asyncio
 
+
 _LOGGER = logging.getLogger(__name__)
 
-class SplitWorkerBase(object, metaclass=abc.ABCMeta):
-    """HttpClient wrapper template."""
+class WorkerBase(object, metaclass=abc.ABCMeta):
+    """Worker template."""
 
     @abc.abstractmethod
     def is_running(self):
@@ -22,7 +23,125 @@ def start(self):
     def stop(self):
         """Stop worker."""
 
-class SplitWorker(SplitWorkerBase):
+class SegmentWorker(WorkerBase):
+    """Segment Worker for processing updates."""
+
+    _centinel = object()
+
+    def __init__(self, synchronize_segment, segment_queue):
+        """
+        Class constructor.
+
+        :param synchronize_segment: handler to perform segment synchronization on incoming event
+        :type synchronize_segment: function
+
+        :param segment_queue: queue with segment updates notifications
+        :type segment_queue: queue
+        """
+        self._segment_queue = segment_queue
+        self._handler = synchronize_segment
+        self._running = False
+        self._worker = None
+
+    def is_running(self):
+        """Return whether the working is running."""
+        return self._running
+
+    def _run(self):
+        """Run worker handler."""
+        while self.is_running():
+            event = self._segment_queue.get()
+            if not self.is_running():
+                break
+            if event == self._centinel:
+                continue
+            _LOGGER.debug('Processing segment_update: %s, change_number: %d',
+                          event.segment_name, event.change_number)
+            try:
+                self._handler(event.segment_name, event.change_number)
+            except Exception:
+                _LOGGER.error('Exception raised in segment synchronization')
+                _LOGGER.debug('Exception information: ', exc_info=True)
+
+    def start(self):
+        """Start worker."""
+        if self.is_running():
+            _LOGGER.debug('Worker is already running')
+            return
+        self._running = True
+
+        _LOGGER.debug('Starting Segment Worker')
+        self._worker = threading.Thread(target=self._run, name='PushSegmentWorker', daemon=True)
+        self._worker.start()
+
+    def stop(self):
+        """Stop worker."""
+        _LOGGER.debug('Stopping Segment Worker')
+        if not self.is_running():
+            _LOGGER.debug('Worker is not running. Ignoring.')
+            return
+        self._running = False
+        self._segment_queue.put(self._centinel)
+
+class SegmentWorkerAsync(WorkerBase):
+    """Segment Worker for processing updates."""
+
+    _centinel = object()
+
+    def __init__(self, synchronize_segment, segment_queue):
+        """
+        Class constructor.
+
+        :param synchronize_segment: handler to perform segment synchronization on incoming event
+        :type synchronize_segment: function
+
+        :param segment_queue: queue with segment updates notifications
+        :type segment_queue: asyncio.Queue
+        """
+        self._segment_queue = segment_queue
+        self._handler = synchronize_segment
+        self._running = False
+
+    def is_running(self):
+        """Return whether the working is running."""
+        return self._running
+
+    async def _run(self):
+        """Run worker handler."""
+        while self.is_running():
+            event = await self._segment_queue.get()
+            if not self.is_running():
+                break
+            if event == self._centinel:
+                continue
+            _LOGGER.debug('Processing segment_update: %s, change_number: %d',
+                          event.segment_name, event.change_number)
+            try:
+                await self._handler(event.segment_name, event.change_number)
+            except Exception:
+                _LOGGER.error('Exception raised in segment synchronization')
+                _LOGGER.debug('Exception information: ', exc_info=True)
+
+    def start(self):
+        """Start worker."""
+        if self.is_running():
+            _LOGGER.debug('Worker is already running')
+            return
+        self._running = True
+
+        _LOGGER.debug('Starting Segment Worker')
+        asyncio.get_event_loop().create_task(self._run())
+
+    async def stop(self):
+        """Stop worker."""
+        _LOGGER.debug('Stopping Segment Worker')
+        if not self.is_running():
+            _LOGGER.debug('Worker is not running. Ignoring.')
+            return
+        self._running = False
+        await self._segment_queue.put(self._centinel)
+
+class SplitWorker(WorkerBase):
     """Feature Flag Worker for processing updates."""
 
     _centinel = object()
@@ -81,7 +200,7 @@ def stop(self):
         self._running = False
         self._feature_flag_queue.put(self._centinel)
 
-class SplitWorkerAsync(SplitWorkerBase):
+class SplitWorkerAsync(WorkerBase):
     """Split Worker for processing updates."""
 
     _centinel = object()

splitio/push/splitworker.py splitio/push/workers.pysplitio/push/splitworker.py renamed to splitio/push/workers.py

@@ -4,8 +4,9 @@
 import pytest
 
 from splitio.api import APIException
-from splitio.push.segmentworker import SegmentWorker
+from splitio.push.workers import SegmentWorker, SegmentWorkerAsync
 from splitio.models.notification import SegmentChangeNotification
+from splitio.optional.loaders import asyncio
 
 change_number_received = None
 segment_name_received = None
@@ -58,3 +59,52 @@ def test_handler(self):
 
         segment_worker.stop()
         assert not segment_worker.is_running()
+
+class SegmentWorkerAsyncTests(object):
+    async def test_on_error(self):
+        q = asyncio.Queue()
+
+        def handler_sync(change_number):
+            raise APIException('some')
+
+        segment_worker = SegmentWorkerAsync(handler_sync, q)
+        segment_worker.start()
+        assert segment_worker.is_running()
+
+        await q.put(SegmentChangeNotification('some', 'SEGMENT_UPDATE', 123456789, 'some'))
+
+        with pytest.raises(Exception):
+            segment_worker._handler()
+
+        assert segment_worker.is_running()
+        assert(self._worker_running())
+        await segment_worker.stop()
+        await asyncio.sleep(.1)
+        assert not segment_worker.is_running()
+        assert(not self._worker_running())
+
+    def _worker_running(self):
+        worker_running = False
+        for task in asyncio.Task.all_tasks():
+            if task._coro.cr_code.co_name == '_run' and not task.done():
+                worker_running = True
+                break
+        return worker_running
+
+    async def test_handler(self):
+        q = asyncio.Queue()
+        segment_worker = SegmentWorkerAsync(handler_sync, q)
+        global change_number_received
+        assert not segment_worker.is_running()
+        segment_worker.start()
+        assert segment_worker.is_running()
+
+        await q.put(SegmentChangeNotification('some', 'SEGMENT_UPDATE', 123456789, 'some'))
+
+        await asyncio.sleep(.1)
+        assert change_number_received == 123456789
+        assert segment_name_received == 'some'
+
+        await segment_worker.stop()
+        await asyncio.sleep(.1)
+        assert(not self._worker_running())

tests/push/test_segment_worker.py

@@ -4,7 +4,7 @@
 import pytest
 
 from splitio.api import APIException
-from splitio.push.splitworker import SplitWorker, SplitWorkerAsync
+from splitio.push.workers import SplitWorker, SplitWorkerAsync
 from splitio.models.notification import SplitChangeNotification
 from splitio.optional.loaders import asyncio