Merge pull request #421 from splitio/async-sync-events

Added sync event async class

Author: chillaq

splitio/sync/event.py

@@ -2,12 +2,13 @@
 import queue
 
 from splitio.api import APIException
-
+from splitio.optional.loaders import asyncio
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class EventSynchronizer(object):
+    """Event Synchronizer class"""
     def __init__(self, events_api, storage, bulk_size):
         """
         Class constructor.
@@ -65,3 +66,64 @@ def synchronize_events(self):
             _LOGGER.error('Exception raised while reporting events')
             _LOGGER.debug('Exception information: ', exc_info=True)
             self._add_to_failed_queue(to_send)
+
+
+class EventSynchronizerAsync(object):
+    """Event Synchronizer async class"""
+    def __init__(self, events_api, storage, bulk_size):
+        """
+        Class constructor.
+
+        :param events_api: Events Api object to send data to the backend
+        :type events_api: splitio.api.events.EventsAPI
+        :param storage: Events Storage
+        :type storage: splitio.storage.EventStorage
+        :param bulk_size: How many events to send per push.
+        :type bulk_size: int
+
+        """
+        self._api = events_api
+        self._event_storage = storage
+        self._bulk_size = bulk_size
+        self._failed = asyncio.Queue()
+
+    async def _get_failed(self):
+        """Return up to <BULK_SIZE> events stored in the failed eventes queue."""
+        events = []
+        count = 0
+        while count < self._bulk_size and self._failed.qsize() > 0:
+            try:
+                events.append(await self._failed.get())
+                count += 1
+            except asyncio.QueueEmpty:
+                # If no more items in queue, break the loop
+                break
+        return events
+
+    async def _add_to_failed_queue(self, events):
+        """
+        Add events that were about to be sent to a secondary queue for failed sends.
+
+        :param events: List of events that failed to be pushed.
+        :type events: list
+        """
+        for event in events:
+            await self._failed.put(event)
+
+    async def synchronize_events(self):
+        """Send events from both the failed and new queues."""
+        to_send = await self._get_failed()
+        if len(to_send) < self._bulk_size:
+            # If the amount of previously failed items is less than the bulk
+            # size, try to complete with new events from storage
+            to_send.extend(await self._event_storage.pop_many(self._bulk_size - len(to_send)))
+
+        if not to_send:
+            return
+
+        try:
+            await self._api.flush_events(to_send)
+        except APIException:
+            _LOGGER.error('Exception raised while reporting events')
+            _LOGGER.debug('Exception information: ', exc_info=True)
+            await self._add_to_failed_queue(to_send)

tests/sync/test_events_synchronizer.py

@@ -8,7 +8,7 @@
 from splitio.api import APIException
 from splitio.storage import EventStorage
 from splitio.models.events import Event
-from splitio.sync.event import EventSynchronizer
+from splitio.sync.event import EventSynchronizer, EventSynchronizerAsync
 
 
 class EventsSynchronizerTests(object):
@@ -66,3 +66,66 @@ def run(x):
         event_synchronizer.synchronize_events()
         assert run._called == 1
         assert event_synchronizer._failed.qsize() == 0
+
+
+class EventsSynchronizerAsyncTests(object):
+    """Events synchronizer async test cases."""
+
+    @pytest.mark.asyncio
+    async def test_synchronize_events_error(self, mocker):
+        storage = mocker.Mock(spec=EventStorage)
+        async def pop_many(*args):
+            return [
+                Event('key1', 'user', 'purchase', 5.3, 123456, None),
+                Event('key2', 'user', 'purchase', 5.3, 123456, None),
+            ]
+        storage.pop_many = pop_many
+
+        api = mocker.Mock()
+        async def run(x):
+            raise APIException("something broke")
+
+        api.flush_events = run
+        event_synchronizer = EventSynchronizerAsync(api, storage, 5)
+        await event_synchronizer.synchronize_events()
+        assert event_synchronizer._failed.qsize() == 2
+
+    @pytest.mark.asyncio
+    async def test_synchronize_events_empty(self, mocker):
+        storage = mocker.Mock(spec=EventStorage)
+        async def pop_many(*args):
+            return []
+        storage.pop_many = pop_many
+
+        api = mocker.Mock()
+        async def run(x):
+            run._called += 1
+
+        run._called = 0
+        api.flush_events = run
+        event_synchronizer = EventSynchronizerAsync(api, storage, 5)
+        await event_synchronizer.synchronize_events()
+        assert run._called == 0
+
+    @pytest.mark.asyncio
+    async def test_synchronize_impressions(self, mocker):
+        storage = mocker.Mock(spec=EventStorage)
+        async def pop_many(*args):
+            return [
+                Event('key1', 'user', 'purchase', 5.3, 123456, None),
+                Event('key2', 'user', 'purchase', 5.3, 123456, None),
+            ]
+        storage.pop_many = pop_many
+
+        api = mocker.Mock()
+        async def run(x):
+            run._called += 1
+            return HttpResponse(200, '', {})
+
+        api.flush_events.side_effect = run
+        run._called = 0
+
+        event_synchronizer = EventSynchronizerAsync(api, storage, 5)
+        await event_synchronizer.synchronize_events()
+        assert run._called == 1
+        assert event_synchronizer._failed.qsize() == 0