-
Notifications
You must be signed in to change notification settings - Fork 13
Expand file tree
/
Copy pathunique_keys_sync.py
More file actions
137 lines (101 loc) · 4.49 KB
/
Copy pathunique_keys_sync.py
File metadata and controls
137 lines (101 loc) · 4.49 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
"""Impressions syncrhonization task."""
import logging
from splitio.tasks import BaseSynchronizationTask
from splitio.tasks.util.asynctask import AsyncTask, AsyncTaskAsync
_LOGGER = logging.getLogger(__name__)
_UNIQUE_KEYS_SYNC_PERIOD = 15 * 60 # 15 minutes
_CLEAR_FILTER_SYNC_PERIOD = 60 * 60 * 24 # 24 hours
class UniqueKeysSyncTaskBase(BaseSynchronizationTask):
"""Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
def start(self):
"""Start executing the unique keys synchronization task."""
self._task.start()
def stop(self, event=None):
"""Stop executing the unique keys 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 unique keys')
self._task.force_execution()
class UniqueKeysSyncTask(UniqueKeysSyncTaskBase):
"""Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
def __init__(self, synchronize_unique_keys, period = _UNIQUE_KEYS_SYNC_PERIOD):
"""
Class constructor.
:param synchronize_unique_keys: sender
:type synchronize_unique_keys: func
:param period: How many seconds to wait between subsequent unique keys pushes to the BE.
:type period: int
"""
self._task = AsyncTask(synchronize_unique_keys, period,
on_stop=synchronize_unique_keys)
def stop(self, event=None):
"""Stop executing the unique keys synchronization task."""
self._task.stop(event)
class UniqueKeysSyncTaskAsync(UniqueKeysSyncTaskBase):
"""Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
def __init__(self, synchronize_unique_keys, period = _UNIQUE_KEYS_SYNC_PERIOD):
"""
Class constructor.
:param synchronize_unique_keys: sender
:type synchronize_unique_keys: func
:param period: How many seconds to wait between subsequent unique keys pushes to the BE.
:type period: int
"""
self._task = AsyncTaskAsync(synchronize_unique_keys, period,
on_stop=synchronize_unique_keys)
async def stop(self):
"""Stop executing the unique keys synchronization task."""
await self._task.stop(True)
class ClearFilterSyncTaskBase(BaseSynchronizationTask):
"""Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
def start(self):
"""Start executing the unique keys synchronization task."""
self._task.start()
def stop(self, event=None):
"""Stop executing the unique keys 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()
class ClearFilterSyncTask(ClearFilterSyncTaskBase):
"""Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
def __init__(self, clear_filter, period = _CLEAR_FILTER_SYNC_PERIOD):
"""
Class constructor.
:param synchronize_unique_keys: sender
:type synchronize_unique_keys: func
:param period: How many seconds to wait between subsequent clearing of bloom filter
:type period: int
"""
self._task = AsyncTask(clear_filter, period,
on_stop=clear_filter)
def stop(self, event=None):
"""Stop executing the unique keys synchronization task."""
self._task.stop(event)
class ClearFilterSyncTaskAsync(ClearFilterSyncTaskBase):
"""Unique Keys synchronization task uses an asynctask.AsyncTask to send MTKs."""
def __init__(self, clear_filter, period = _CLEAR_FILTER_SYNC_PERIOD):
"""
Class constructor.
:param synchronize_unique_keys: sender
:type synchronize_unique_keys: func
:param period: How many seconds to wait between subsequent clearing of bloom filter
:type period: int
"""
self._task = AsyncTaskAsync(clear_filter, period,
on_stop=clear_filter)
async def stop(self):
"""Stop executing the unique keys synchronization task."""
await self._task.stop(True)