Forced async tasks to wait for completion

Author: chillaq

splitio/tasks/events_sync.py

@@ -73,4 +73,4 @@ def __init__(self, synchronize_events, period):
 
     async def stop(self, event=None):
         """Stop executing the events synchronization task."""
-        await self._task.stop()
+        await self._task.stop(True)

splitio/tasks/impressions_sync.py

@@ -75,7 +75,7 @@ def __init__(self, synchronize_impressions, period):
 
     async def stop(self, event=None):
         """Stop executing the impressions synchronization task."""
-        await self._task.stop()
+        await self._task.stop(True)
 
 
 class ImpressionsCountSyncTaskBase(BaseSynchronizationTask):
@@ -136,6 +136,6 @@ def __init__(self, synchronize_counters):
         """
         self._task = AsyncTaskAsync(synchronize_counters, self._PERIOD, on_stop=synchronize_counters)
 
-    async def stop(self, event=None):
+    async def stop(self):
         """Stop executing the impressions synchronization task."""
-        await self._task.stop()
+        await self._task.stop(True)

splitio/tasks/segment_sync.py

@@ -60,6 +60,6 @@ def __init__(self, synchronize_segments, period):
         """
         self._task = asynctask.AsyncTaskAsync(synchronize_segments, period, on_init=None)
 
-    async def stop(self, event=None):
+    async def stop(self):
         """Stop segment synchronization."""
-        await self._task.stop(event)
+        await self._task.stop(True)

splitio/tasks/split_sync.py

@@ -66,4 +66,4 @@ def __init__(self, synchronize_splits, period):
 
     async def stop(self, event=None):
         """Stop the task. Accept an optional event to set when the task has finished."""
-        await self._task.stop()
+        await self._task.stop(True)

splitio/tasks/telemetry_sync.py

@@ -2,10 +2,36 @@
 import logging
 
 from splitio.tasks import BaseSynchronizationTask
-from splitio.tasks.util.asynctask import AsyncTask
+from splitio.tasks.util.asynctask import AsyncTask, AsyncTaskAsync
 
 _LOGGER = logging.getLogger(__name__)
 
+class TelemetrySyncTaskBase(BaseSynchronizationTask):
+    """Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
+
+    def start(self):
+        """Start executing the telemetry synchronization task."""
+        self._task.start()
+
+    def stop(self, event=None):
+        """Stop executing the unique telemetry synchronization task."""
+        pass
+
+    def is_running(self):
+        """
+        Return whether the task is running or not.
+
+        :return: True if the task is running. False otherwise.
+        :rtype: bool
+        """
+        return self._task.running()
+
+    def flush(self):
+        """Flush unique keys."""
+        _LOGGER.debug('Forcing flush execution for telemetry')
+        self._task.force_execution()
+
+
 class TelemetrySyncTask(BaseSynchronizationTask):
     """Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
 
@@ -22,24 +48,27 @@ def __init__(self, synchronize_telemetry, period):
         self._task = AsyncTask(synchronize_telemetry, period,
                                on_stop=synchronize_telemetry)
 
-    def start(self):
-        """Start executing the telemetry synchronization task."""
-        self._task.start()
-
     def stop(self, event=None):
         """Stop executing the unique telemetry synchronization task."""
         self._task.stop(event)
 
-    def is_running(self):
+
+class TelemetrySyncTaskAsync(BaseSynchronizationTask):
+    """Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
+
+    def __init__(self, synchronize_telemetry, period):
         """
-        Return whether the task is running or not.
+        Class constructor.
 
-        :return: True if the task is running. False otherwise.
-        :rtype: bool
+        :param synchronize_telemetry: sender
+        :type synchronize_telemetry: func
+        :param period: How many seconds to wait between subsequent unique keys pushes to the BE.
+        :type period: int
         """
-        return self._task.running()
 
-    def flush(self):
-        """Flush unique keys."""
-        _LOGGER.debug('Forcing flush execution for telemetry')
-        self._task.force_execution()
+        self._task = AsyncTaskAsync(synchronize_telemetry, period,
+                               on_stop=synchronize_telemetry)
+
+    async def stop(self):
+        """Stop executing the unique telemetry synchronization task."""
+        await self._task.stop(True)

splitio/tasks/unique_keys_sync.py

@@ -71,9 +71,9 @@ def __init__(self, synchronize_unique_keys, period = _UNIQUE_KEYS_SYNC_PERIOD):
         self._task = AsyncTaskAsync(synchronize_unique_keys, period,
                                on_stop=synchronize_unique_keys)
 
-    async def stop(self, event=None):
+    async def stop(self):
         """Stop executing the unique keys synchronization task."""
-        await self._task.stop(event)
+        await self._task.stop(True)
 
 
 class ClearFilterSyncTaskBase(BaseSynchronizationTask):
@@ -123,6 +123,6 @@ def __init__(self, clear_filter, period = _CLEAR_FILTER_SYNC_PERIOD):
         self._task = AsyncTaskAsync(clear_filter, period,
                                on_stop=clear_filter)
 
-    async def stop(self, event=None):
+    async def stop(self):
         """Stop executing the unique keys synchronization task."""
-        await self._task.stop(event)
+        await self._task.stop(True)