Merge pull request #422 from splitio/async-sync-imps

added impressions and impressions count sync classes

Author: chillaq

splitio/sync/impression.py

@@ -2,11 +2,13 @@
 import queue
 
 from splitio.api import APIException
+from splitio.optional.loaders import asyncio
 
 _LOGGER = logging.getLogger(__name__)
 
 
 class ImpressionSynchronizer(object):
+    """Impressions synchronizer class."""
     def __init__(self, impressions_api, storage, bulk_size):
         """
         Class constructor.
@@ -95,3 +97,95 @@ def synchronize_counters(self):
         except APIException:
             _LOGGER.error('Exception raised while reporting impression counts')
             _LOGGER.debug('Exception information: ', exc_info=True)
+
+
+class ImpressionSynchronizerAsync(object):
+    """Impressions async synchronizer class."""
+    def __init__(self, impressions_api, storage, bulk_size):
+        """
+        Class constructor.
+
+        :param impressions_api: Impressions Api object to send data to the backend
+        :type impressions_api: splitio.api.impressions.ImpressionsAPI
+        :param storage: Impressions Storage
+        :type storage: splitio.storage.ImpressionsStorage
+        :param bulk_size: How many impressions to send per push.
+        :type bulk_size: int
+
+        """
+        self._api = impressions_api
+        self._impression_storage = storage
+        self._bulk_size = bulk_size
+        self._failed = asyncio.Queue()
+
+    async def _get_failed(self):
+        """Return up to <BULK_SIZE> impressions stored in the failed impressions queue."""
+        imps = []
+        count = 0
+        while count < self._bulk_size and self._failed.qsize() > 0:
+            try:
+                imps.append(await self._failed.get())
+                count += 1
+            except asyncio.QueueEmpty:
+                # If no more items in queue, break the loop
+                break
+        return imps
+
+    async def _add_to_failed_queue(self, imps):
+        """
+        Add impressions that were about to be sent to a secondary queue for failed sends.
+
+        :param imps: List of impressions that failed to be pushed.
+        :type imps: list
+        """
+        for impression in imps:
+            await self._failed.put(impression)
+
+    async def synchronize_impressions(self):
+        """Send impressions 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 impressions from storage
+            to_send.extend(await self._impression_storage.pop_many(self._bulk_size - len(to_send)))
+
+        if not to_send:
+            return
+
+        try:
+            await self._api.flush_impressions(to_send)
+        except APIException:
+            _LOGGER.error('Exception raised while reporting impressions')
+            _LOGGER.debug('Exception information: ', exc_info=True)
+            await self._add_to_failed_queue(to_send)
+
+
+class ImpressionsCountSynchronizerAsync(object):
+    def __init__(self, impressions_api, imp_counter):
+        """
+        Class constructor.
+
+        :param impressions_api: Impressions Api object to send data to the backend
+        :type impressions_api: splitio.api.impressions.ImpressionsAPI
+        :param impressions_manager: Impressions manager instance
+        :type impressions_manager: splitio.engine.impressions.Manager
+
+        """
+        self._impressions_api = impressions_api
+        self._impressions_counter = imp_counter
+
+    async def synchronize_counters(self):
+        """Send impressions from both the failed and new queues."""
+
+        if self._impressions_counter == None:
+            return
+
+        to_send = await self._impressions_counter.pop_all()
+        if not to_send:
+            return
+
+        try:
+            await self._impressions_api.flush_counters(to_send)
+        except APIException:
+            _LOGGER.error('Exception raised while reporting impression counts')
+            _LOGGER.debug('Exception information: ', exc_info=True)

tests/sync/test_impressions_count_synchronizer.py

@@ -9,7 +9,7 @@
 from splitio.engine.impressions.impressions import Manager as ImpressionsManager
 from splitio.engine.impressions.manager import Counter
 from splitio.engine.impressions.strategies import StrategyOptimizedMode
-from splitio.sync.impression import ImpressionsCountSynchronizer
+from splitio.sync.impression import ImpressionsCountSynchronizer, ImpressionsCountSynchronizerAsync
 from splitio.api.impressions import ImpressionsAPI
 
 
@@ -36,3 +36,40 @@ def test_synchronize_impressions_counts(self, mocker):
         assert api.flush_counters.mock_calls[0] == mocker.call(counters)
 
         assert len(api.flush_counters.mock_calls) == 1
+
+
+class ImpressionsCountSynchronizerAsyncTests(object):
+    """ImpressionsCount synchronizer test cases."""
+
+    @pytest.mark.asyncio
+    async def test_synchronize_impressions_counts(self, mocker):
+        counter = mocker.Mock(spec=Counter)
+
+        self.called = 0
+        async def pop_all():
+            self.called += 1
+            return [
+                Counter.CountPerFeature('f1', 123, 2),
+                Counter.CountPerFeature('f2', 123, 123),
+                Counter.CountPerFeature('f1', 456, 111),
+                Counter.CountPerFeature('f2', 456, 222)
+            ]
+        counter.pop_all = pop_all
+
+        self.counters = None
+        async def flush_counters(counters):
+            self.counters = counters
+            return HttpResponse(200, '', {})
+        api = mocker.Mock(spec=ImpressionsAPI)
+        api.flush_counters = flush_counters
+
+        impression_count_synchronizer = ImpressionsCountSynchronizerAsync(api, counter)
+        await impression_count_synchronizer.synchronize_counters()
+
+        assert self.counters == [
+                Counter.CountPerFeature('f1', 123, 2),
+                Counter.CountPerFeature('f2', 123, 123),
+                Counter.CountPerFeature('f1', 456, 111),
+                Counter.CountPerFeature('f2', 456, 222)
+            ]
+        assert self.called == 1

tests/sync/test_impressions_synchronizer.py

@@ -8,7 +8,7 @@
 from splitio.api import APIException
 from splitio.storage import ImpressionStorage
 from splitio.models.impressions import Impression
-from splitio.sync.impression import ImpressionSynchronizer
+from splitio.sync.impression import ImpressionSynchronizer, ImpressionSynchronizerAsync
 
 
 class ImpressionsSynchronizerTests(object):
@@ -66,3 +66,68 @@ def run(x):
         impression_synchronizer.synchronize_impressions()
         assert run._called == 1
         assert impression_synchronizer._failed.qsize() == 0
+
+
+class ImpressionsSynchronizerAsyncTests(object):
+    """Impressions synchronizer test cases."""
+
+    @pytest.mark.asyncio
+    async def test_synchronize_impressions_error(self, mocker):
+        storage = mocker.Mock(spec=ImpressionStorage)
+        async def pop_many(*args):
+            return [
+                Impression('key1', 'split1', 'on', 'l1', 123456, 'b1', 321654),
+                Impression('key2', 'split1', 'on', 'l1', 123456, 'b1', 321654),
+            ]
+        storage.pop_many = pop_many
+        api = mocker.Mock()
+
+        async def run(x):
+            raise APIException("something broke")
+        api.flush_impressions = run
+
+        impression_synchronizer = ImpressionSynchronizerAsync(api, storage, 5)
+        await impression_synchronizer.synchronize_impressions()
+        assert impression_synchronizer._failed.qsize() == 2
+
+    @pytest.mark.asyncio
+    async def test_synchronize_impressions_empty(self, mocker):
+        storage = mocker.Mock(spec=ImpressionStorage)
+        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_impressions = run
+        impression_synchronizer = ImpressionSynchronizerAsync(api, storage, 5)
+        await impression_synchronizer.synchronize_impressions()
+        assert run._called == 0
+
+    @pytest.mark.asyncio
+    async def test_synchronize_impressions(self, mocker):
+        storage = mocker.Mock(spec=ImpressionStorage)
+        async def pop_many(*args):
+            return [
+                Impression('key1', 'split1', 'on', 'l1', 123456, 'b1', 321654),
+                Impression('key2', 'split1', 'on', 'l1', 123456, 'b1', 321654),
+            ]
+        storage.pop_many = pop_many
+
+        api = mocker.Mock()
+
+        async def run(x):
+            run._called += 1
+            return HttpResponse(200, '', {})
+
+        api.flush_impressions = run
+        run._called = 0
+
+        impression_synchronizer = ImpressionSynchronizerAsync(api, storage, 5)
+        await impression_synchronizer.synchronize_impressions()
+        assert run._called == 1
+        assert impression_synchronizer._failed.qsize() == 0