Merge pull request #382 from splitio/async-splitworker

Async splitworker

Author: chillaq

splitio/api/client.py

@@ -4,16 +4,7 @@
 import urllib
 import abc
 
-try:
-    import aiohttp
-except ImportError:
-    def missing_asyncio_dependencies(*_, **__):
-        """Fail if missing dependencies are used."""
-        raise NotImplementedError(
-            'Missing aiohttp dependency. '
-            'Please use `pip install splitio_client[asyncio]` to install the sdk with asyncio support'
-        )
-    aiohttp = missing_asyncio_dependencies
+from splitio.optional.loaders import aiohttp
 
 SDK_URL = 'https://sdk.split.io/api'
 EVENTS_URL = 'https://events.split.io/api'
@@ -256,5 +247,5 @@ async def post(self, server, path, apikey, body, query=None, extra_headers=None)
             ) as response:
                 body = await response.text()
                 return HttpResponse(response.status, body, response.headers)
-        except Exception as exc:  # pylint: disable=broad-except
+        except aiohttp.ClientError as exc:  # pylint: disable=broad-except
             raise HttpClientException('aiohttp library is throwing exceptions') from exc

splitio/optional/__init__.py

@@ -0,0 +1,12 @@
+try:
+    import asyncio
+    import aiohttp
+except ImportError:
+    def missing_asyncio_dependencies(*_, **__):
+        """Fail if missing dependencies are used."""
+        raise NotImplementedError(
+            'Missing aiohttp dependency. '
+            'Please use `pip install splitio_client[asyncio]` to install the sdk with asyncio support'
+        )
+    aiohttp = missing_asyncio_dependencies
+    asyncio = missing_asyncio_dependencies

splitio/optional/loaders.py

@@ -1,12 +1,28 @@
 """Feature Flag 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 SplitWorker(object):
+    @abc.abstractmethod
+    def is_running(self):
+        """Return whether the working is running."""
+
+    @abc.abstractmethod
+    def start(self):
+        """Start worker."""
+
+    @abc.abstractmethod
+    def stop(self):
+        """Stop worker."""
+
+class SplitWorker(SplitWorkerBase):
     """Feature Flag Worker for processing updates."""
 
     _centinel = object()
@@ -64,3 +80,62 @@ def stop(self):
             return
         self._running = False
         self._feature_flag_queue.put(self._centinel)
+
+class SplitWorkerAsync(SplitWorkerBase):
+    """Split Worker for processing updates."""
+
+    _centinel = object()
+
+    def __init__(self, synchronize_split, split_queue):
+        """
+        Class constructor.
+
+        :param synchronize_split: handler to perform split synchronization on incoming event
+        :type synchronize_split: callable
+
+        :param split_queue: queue with split updates notifications
+        :type split_queue: queue
+        """
+        self._split_queue = split_queue
+        self._handler = synchronize_split
+        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():
+            _LOGGER.error("_run")
+            event = await self._split_queue.get()
+            if not self.is_running():
+                break
+            if event == self._centinel:
+                continue
+            _LOGGER.debug('Processing split_update %d', event.change_number)
+            try:
+                _LOGGER.error(event.change_number)
+                await self._handler(event.change_number)
+            except Exception:  # pylint: disable=broad-except
+                _LOGGER.error('Exception raised in split 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 Split Worker')
+        asyncio.get_event_loop().create_task(self._run())
+
+    async def stop(self):
+        """Stop worker."""
+        _LOGGER.debug('Stopping Split Worker')
+        if not self.is_running():
+            _LOGGER.debug('Worker is not running')
+            return
+        self._running = False
+        await self._split_queue.put(self._centinel)

splitio/push/splitworker.py

@@ -166,7 +166,7 @@ async def test_get(self, mocker):
         response_mock = MockResponse('ok', 200, {})
         get_mock = mocker.Mock()
         get_mock.return_value = response_mock
-        mocker.patch('splitio.api.client.aiohttp.ClientSession.get', new=get_mock)
+        mocker.patch('splitio.optional.loaders.aiohttp.ClientSession.get', new=get_mock)
         httpclient = client.HttpClientAsync()
         response = await httpclient.get('sdk', 'test1', 'some_api_key', {'param1': 123}, {'h1': 'abc'})
         assert response.status_code == 200
@@ -197,7 +197,7 @@ async def test_get_custom_urls(self, mocker):
         response_mock = MockResponse('ok', 200, {})
         get_mock = mocker.Mock()
         get_mock.return_value = response_mock
-        mocker.patch('splitio.api.client.aiohttp.ClientSession.get', new=get_mock)
+        mocker.patch('splitio.optional.loaders.aiohttp.ClientSession.get', new=get_mock)
         httpclient = client.HttpClientAsync(sdk_url='https://sdk.com', events_url='https://events.com')
         response = await httpclient.get('sdk', 'test1', 'some_api_key', {'param1': 123}, {'h1': 'abc'})
         call = mocker.call(
@@ -228,7 +228,7 @@ async def test_post(self, mocker):
         response_mock = MockResponse('ok', 200, {})
         get_mock = mocker.Mock()
         get_mock.return_value = response_mock
-        mocker.patch('splitio.api.client.aiohttp.ClientSession.post', new=get_mock)
+        mocker.patch('splitio.optional.loaders.aiohttp.ClientSession.post', new=get_mock)
         httpclient = client.HttpClientAsync()
         response = await httpclient.post('sdk', 'test1', 'some_api_key', {'p1': 'a'}, {'param1': 123}, {'h1': 'abc'})
         call = mocker.call(
@@ -260,7 +260,7 @@ async def test_post_custom_urls(self, mocker):
         response_mock = MockResponse('ok', 200, {})
         get_mock = mocker.Mock()
         get_mock.return_value = response_mock
-        mocker.patch('splitio.api.client.aiohttp.ClientSession.post', new=get_mock)
+        mocker.patch('splitio.optional.loaders.aiohttp.ClientSession.post', new=get_mock)
         httpclient = client.HttpClientAsync(sdk_url='https://sdk.com', events_url='https://events.com')
         response = await httpclient.post('sdk', 'test1', 'some_api_key', {'p1': 'a'}, {'param1': 123}, {'h1': 'abc'})
         call = mocker.call(

tests/api/test_httpclient.py

@@ -4,8 +4,9 @@
 import pytest
 
 from splitio.api import APIException
-from splitio.push.splitworker import SplitWorker
+from splitio.push.splitworker import SplitWorker, SplitWorkerAsync
 from splitio.models.notification import SplitChangeNotification
+from splitio.optional.loaders import asyncio
 
 change_number_received = None
 
@@ -15,6 +16,11 @@ def handler_sync(change_number):
     change_number_received = change_number
     return
 
+async def handler_async(change_number):
+    global change_number_received
+    change_number_received = change_number
+    return
+
 
 class SplitWorkerTests(object):
 
@@ -55,3 +61,57 @@ def test_handler(self):
 
         split_worker.stop()
         assert not split_worker.is_running()
+
+class SplitWorkerAsyncTests(object):
+
+    async def test_on_error(self):
+        q = asyncio.Queue()
+
+        def handler_sync(change_number):
+            raise APIException('some')
+
+        split_worker = SplitWorkerAsync(handler_sync, q)
+        split_worker.start()
+        assert split_worker.is_running()
+
+        await q.put(SplitChangeNotification('some', 'SPLIT_UPDATE', 123456789))
+        with pytest.raises(Exception):
+            split_worker._handler()
+
+        assert split_worker.is_running()
+        assert(self._worker_running())
+
+        await split_worker.stop()
+        await asyncio.sleep(.1)
+
+        assert not split_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()
+        split_worker = SplitWorkerAsync(handler_async, q)
+
+        assert not split_worker.is_running()
+        split_worker.start()
+        assert split_worker.is_running()
+        assert(self._worker_running())
+
+        global change_number_received
+        await q.put(SplitChangeNotification('some', 'SPLIT_UPDATE', 123456789))
+        await asyncio.sleep(1)
+
+        assert change_number_received == 123456789
+
+        await split_worker.stop()
+        await asyncio.sleep(.1)
+
+        assert not split_worker.is_running()
+        assert(not self._worker_running())