Merge pull request #388 from splitio/async-processor

Added processor async class

Author: chillaq

splitio/push/processor.py

@@ -1,13 +1,28 @@
 """Message processor & Notification manager keeper implementations."""
 
 from queue import Queue
+import abc
 
 from splitio.push.parser import UpdateType
-from splitio.push.workers import SplitWorker
-from splitio.push.workers import SegmentWorker
+from splitio.push.workers import SplitWorker, SplitWorkerAsync, SegmentWorker, SegmentWorkerAsync
+from splitio.optional.loaders import asyncio
 
+class MessageProcessorBase(object, metaclass=abc.ABCMeta):
+    """Message processor template."""
 
-class MessageProcessor(object):
+    @abc.abstractmethod
+    def update_workers_status(self, enabled):
+        """Enable/Disable push update workers."""
+
+    @abc.abstractmethod
+    def handle(self, event):
+        """Handle incoming update event."""
+
+    @abc.abstractmethod
+    def shutdown(self):
+        """Stop splits & segments workers."""
+
+class MessageProcessor(MessageProcessorBase):
     """Message processor class."""
 
     def __init__(self, synchronizer):
@@ -89,3 +104,87 @@ def shutdown(self):
         """Stop splits & segments workers."""
         self._split_worker.stop()
         self._segments_worker.stop()
+
+
+class MessageProcessorAsync(MessageProcessorBase):
+    """Message processor class."""
+
+    def __init__(self, synchronizer):
+        """
+        Class constructor.
+
+        :param synchronizer: synchronizer component
+        :type synchronizer: splitio.sync.synchronizer.Synchronizer
+        """
+        self._split_queue = asyncio.Queue()
+        self._segments_queue = asyncio.Queue()
+        self._synchronizer = synchronizer
+        self._split_worker = SplitWorkerAsync(synchronizer.synchronize_splits, self._split_queue)
+        self._segments_worker = SegmentWorkerAsync(synchronizer.synchronize_segment, self._segments_queue)
+        self._handlers = {
+            UpdateType.SPLIT_UPDATE: self._handle_split_update,
+            UpdateType.SPLIT_KILL: self._handle_split_kill,
+            UpdateType.SEGMENT_UPDATE: self._handle_segment_change
+        }
+
+    async def _handle_split_update(self, event):
+        """
+        Handle incoming split update notification.
+
+        :param event: Incoming split change event
+        :type event: splitio.push.parser.SplitChangeUpdate
+        """
+        await self._split_queue.put(event)
+
+    async def _handle_split_kill(self, event):
+        """
+        Handle incoming split kill notification.
+
+        :param event: Incoming split kill event
+        :type event: splitio.push.parser.SplitKillUpdate
+        """
+        await self._synchronizer.kill_split(event.split_name, event.default_treatment,
+                                      event.change_number)
+        await self._split_queue.put(event)
+
+    async def _handle_segment_change(self, event):
+        """
+        Handle incoming segment update notification.
+
+        :param event: Incoming segment change event
+        :type event: splitio.push.parser.Update
+        """
+        await self._segments_queue.put(event)
+
+    async def update_workers_status(self, enabled):
+        """
+        Enable/Disable push update workers.
+
+        :param enabled: if True, enable workers. If False, disable them.
+        :type enabled: bool
+        """
+        if enabled:
+            self._split_worker.start()
+            self._segments_worker.start()
+        else:
+            await self._split_worker.stop()
+            await self._segments_worker.stop()
+
+    async def handle(self, event):
+        """
+        Handle incoming update event.
+
+        :param event: incoming data update event.
+        :type event: splitio.push.BaseUpdate
+        """
+        try:
+            handle = self._handlers[event.update_type]
+        except KeyError as exc:
+            raise Exception('no handler for notification type: %s' % event.update_type) from exc
+
+        await handle(event)
+
+    async def shutdown(self):
+        """Stop splits & segments workers."""
+        await self._split_worker.stop()
+        await self._segments_worker.stop()

splitio/push/workers.py

@@ -130,7 +130,7 @@ def start(self):
         self._running = True
 
         _LOGGER.debug('Starting Segment Worker')
-        asyncio.get_event_loop().create_task(self._run())
+        asyncio.get_running_loop().create_task(self._run())
 
     async def stop(self):
         """Stop worker."""
@@ -248,7 +248,7 @@ def start(self):
         self._running = True
 
         _LOGGER.debug('Starting Split Worker')
-        asyncio.get_event_loop().create_task(self._run())
+        asyncio.get_running_loop().create_task(self._run())
 
     async def stop(self):
         """Stop worker."""

tests/push/test_processor.py

@@ -1,8 +1,11 @@
 """Message processor tests."""
 from queue import Queue
-from splitio.push.processor import MessageProcessor
-from splitio.sync.synchronizer import Synchronizer
+import pytest
+
+from splitio.push.processor import MessageProcessor, MessageProcessorAsync
+from splitio.sync.synchronizer import Synchronizer, SynchronizerAsync
 from splitio.push.parser import SplitChangeUpdate, SegmentChangeUpdate, SplitKillUpdate
+from splitio.optional.loaders import asyncio
 
 
 class ProcessorTests(object):
@@ -56,3 +59,59 @@ def test_segment_change(self, mocker):
     def test_todo(self):
         """Fix previous tests so that we validate WHICH queue the update is pushed into."""
         assert NotImplementedError("DO THAT")
+
+class ProcessorAsyncTests(object):
+    """Message processor test cases."""
+
+    @pytest.mark.asyncio
+    async def test_split_change(self, mocker):
+        """Test split change is properly handled."""
+        sync_mock = mocker.Mock(spec=Synchronizer)
+        self._update = None
+        async def put_mock(first, event):
+            self._update = event
+
+        mocker.patch('splitio.push.processor.asyncio.Queue.put', new=put_mock)
+        processor = MessageProcessorAsync(sync_mock)
+        update = SplitChangeUpdate('sarasa', 123, 123)
+        await processor.handle(update)
+        assert update == self._update
+
+    @pytest.mark.asyncio
+    async def test_split_kill(self, mocker):
+        """Test split kill is properly handled."""
+
+        self._killed_split = None
+        async def kill_mock(se, split_name, default_treatment, change_number):
+            self._killed_split = (split_name, default_treatment, change_number)
+
+        mocker.patch('splitio.sync.synchronizer.SynchronizerAsync.kill_split', new=kill_mock)
+        sync_mock = SynchronizerAsync()
+
+        self._update = None
+        async def put_mock(first, event):
+            self._update = event
+
+        mocker.patch('splitio.push.processor.asyncio.Queue.put', new=put_mock)
+        processor = MessageProcessorAsync(sync_mock)
+        update = SplitKillUpdate('sarasa', 123, 456, 'some_split', 'off')
+        await processor.handle(update)
+        assert update == self._update
+        assert ('some_split', 'off', 456) == self._killed_split
+
+    @pytest.mark.asyncio
+    async def test_segment_change(self, mocker):
+        """Test segment change is properly handled."""
+
+        sync_mock = SynchronizerAsync()
+        queue_mock = mocker.Mock(spec=asyncio.Queue)
+
+        self._update = None
+        async def put_mock(first, event):
+            self._update = event
+
+        mocker.patch('splitio.push.processor.asyncio.Queue.put', new=put_mock)
+        processor = MessageProcessorAsync(sync_mock)
+        update = SegmentChangeUpdate('sarasa', 123, 123, 'some_segment')
+        await processor.handle(update)
+        assert update == self._update