Skip to content

Feature/sdk events #620

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
chillaq merged 38 commits into from
Jan 27, 2026
Merged

Feature/sdk events #620

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
38 commits
Select commit Hold shift + click to select a range
c65fd01
added models, events config data and events metadata
chillaq Jan 7, 2026
661d248
updated metadata to recent spec
chillaq Jan 8, 2026
5d814b5
added events manager and events delivery
chillaq Jan 9, 2026
a37d3b9
added internal sdk task
chillaq Jan 12, 2026
e019bb3
polish
chillaq Jan 13, 2026
a5aa01d
Merge pull request #608 from splitio/FME-12219-sdk-events-models
chillaq Jan 13, 2026
59a5530
polish
chillaq Jan 13, 2026
b92badd
Merge pull request #610 from splitio/FME-12220-sdk-events-manager
chillaq Jan 13, 2026
7c91986
polish
chillaq Jan 13, 2026
27792de
Merge pull request #611 from splitio/FME-12221-sdk-event-task
chillaq Jan 13, 2026
6e3ea36
Updated split storage
chillaq Jan 13, 2026
0989330
Merge pull request #612 from splitio/FME-12222-sdk-events-split-storage
chillaq Jan 13, 2026
b00410d
updated segments and rb segments storages
chillaq Jan 14, 2026
00227ce
Merge pull request #613 from splitio/FME-12223-sdk-events-segments
chillaq Jan 14, 2026
8171606
update factory class for ready and timedout events
chillaq Jan 14, 2026
6d344a6
updated client and factory classes
chillaq Jan 16, 2026
c14e651
Merge pull request #614 from splitio/FME-12227-sdk-events-factory
chillaq Jan 16, 2026
02027af
Merge pull request #615 from splitio/FME-12224-sdk-event-client
chillaq Jan 16, 2026
f0d85ba
updated sdk ready firing after subscription and integration tests
chillaq Jan 16, 2026
983a740
avoid fire events if no items added or removed from storage
chillaq Jan 20, 2026
0df687f
Merge pull request #616 from splitio/FME-12225-sdk-events-integration
chillaq Jan 20, 2026
7194c0a
added async classes
chillaq Jan 21, 2026
f2ad152
finish tests
chillaq Jan 21, 2026
4327a30
updated localhost classes and tests
chillaq Jan 22, 2026
e7f721a
fixed typo for segment event type
chillaq Jan 22, 2026
b92aa3c
Merge branch 'feature/sdk-events' into FME-12276-async-classes
chillaq Jan 22, 2026
11d56fd
fixed typo for segment update type
chillaq Jan 22, 2026
f2608fa
Merge branch 'FME-12276-async-classes' into FME-12226-events-localhost
chillaq Jan 22, 2026
e06630f
Merge pull request #617 from splitio/FME-12276-async-classes
chillaq Jan 22, 2026
06950f0
Merge pull request #618 from splitio/FME-12226-events-localhost
chillaq Jan 22, 2026
8f29ba9
ignored fetching rbs if list is empty
chillaq Jan 22, 2026
3849e9b
Merge pull request #619 from splitio/FME-12322-fix-redis-prefix
chillaq Jan 22, 2026
09dd7e5
removed name param from creat_task, supported only after 3.8
chillaq Jan 22, 2026
15ca507
updated changes
chillaq Jan 27, 2026
27e2592
fixed test
chillaq Jan 27, 2026
2244773
polishing
chillaq Jan 27, 2026
3a9d2a7
fixed calling fire event function
chillaq Jan 27, 2026
43dfb55
updated license and added notice
chillaq Jan 27, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Prev Previous commit
Next Next commit
update factory class for ready and timedout events
  • Loading branch information
@chillaq
chillaq committed Jan 14, 2026
commit 817160685845b0147706e9741195842ca7e62c0c
33 changes: 25 additions & 8 deletions splitio/client/factory.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -20,6 +20,10 @@
from splitio.engine.impressions.manager import Counter as ImpressionsCounter
from splitio.engine.impressions.unique_keys_tracker import UniqueKeysTracker, UniqueKeysTrackerAsync
from splitio.models.fallback_config import FallbackTreatmentCalculator
from splitio.events.events_metadata import EventsMetadata, SdkEventType
from splitio.models.notification import SdkInternalEventNotification
from splitio.models.events import SdkInternalEvent

# Storage
from splitio.storage.inmemmory import InMemorySplitStorage, InMemorySegmentStorage, \
InMemoryImpressionStorage, InMemoryEventStorage, InMemoryTelemetryStorage, LocalhostTelemetryStorage, \
Expand Down Expand Up @@ -166,6 +170,7 @@ def __init__( # pylint: disable=too-many-arguments
storages,
labels_enabled,
recorder,
internal_events_queue,
sync_manager=None,
sdk_ready_flag=None,
telemetry_producer=None,
Expand Down Expand Up @@ -204,6 +209,7 @@ def __init__( # pylint: disable=too-many-arguments
_LOGGER.debug("Running in threading mode")
self._sdk_internal_ready_flag = sdk_ready_flag
self._fallback_treatment_calculator = fallback_treatment_calculator
self._internal_events_queue = internal_events_queue
self._start_status_updater()

def _start_status_updater(self):
Expand All @@ -224,12 +230,15 @@ def _start_status_updater(self):
ready_updater.start()
else:
self._status = Status.READY

self._internal_events_queue.put(SdkInternalEventNotification(SdkInternalEvent.SDK_READY, None))

def _update_status_when_ready(self):
"""Wait until the sdk is ready and update the status."""
self._sdk_internal_ready_flag.wait()
self._status = Status.READY
self._sdk_ready_flag.set()
self._internal_events_queue.put(SdkInternalEventNotification(SdkInternalEvent.SDK_READY, None))

self._telemetry_init_producer.record_ready_time(get_current_epoch_time_ms() - self._ready_time)
redundant_factory_count, active_factory_count = _get_active_and_redundant_count()
self._telemetry_init_producer.record_active_and_redundant_factories(active_factory_count, redundant_factory_count)
Expand Down Expand Up @@ -270,6 +279,7 @@ def block_until_ready(self, timeout=None):

if not ready:
self._telemetry_init_producer.record_bur_time_out()
self._internal_events_queue.put(SdkInternalEventNotification(SdkInternalEvent.SDK_TIMED_OUT, None))
raise TimeoutException('SDK Initialization: time of %d exceeded' % timeout)

def destroy(self, destroyed_event=None):
Expand Down Expand Up @@ -548,11 +558,11 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None, # pyl
'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata, telemetry_runtime_producer),
}

events_queue = queue.Queue()
internal_events_queue = queue.Queue()
storages = {
'splits': InMemorySplitStorage(events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
'segments': InMemorySegmentStorage(events_queue),
'rule_based_segments': InMemoryRuleBasedSegmentStorage(events_queue),
'splits': InMemorySplitStorage(internal_events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
'segments': InMemorySegmentStorage(internal_events_queue),
'rule_based_segments': InMemoryRuleBasedSegmentStorage(internal_events_queue),
'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize'], telemetry_runtime_producer),
'events': InMemoryEventStorage(cfg['eventsQueueSize'], telemetry_runtime_producer),
}
Expand Down Expand Up @@ -629,14 +639,14 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None, # pyl
synchronizer._split_synchronizers._segment_sync.shutdown()

return SplitFactory(api_key, storages, cfg['labelsEnabled'],
recorder, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization,
recorder, internal_events_queue, manager, None, telemetry_producer, telemetry_init_producer, telemetry_submitter, preforked_initialization=preforked_initialization,
fallback_treatment_calculator=FallbackTreatmentCalculator(cfg['fallbackTreatments']))

initialization_thread = threading.Thread(target=manager.start, name="SDKInitializer", daemon=True)
initialization_thread.start()

return SplitFactory(api_key, storages, cfg['labelsEnabled'],
recorder, manager, sdk_ready_flag,
recorder, internal_events_queue, manager, sdk_ready_flag,
telemetry_producer, telemetry_init_producer,
telemetry_submitter, fallback_treatment_calculator = FallbackTreatmentCalculator(cfg['fallbackTreatments']))

Expand Down Expand Up @@ -826,12 +836,14 @@ def _build_redis_factory(api_key, cfg):
initialization_thread.start()

telemetry_init_producer.record_config(cfg, {}, 0, 0)

internal_events_queue = queue.Queue()

split_factory = SplitFactory(
api_key,
storages,
cfg['labelsEnabled'],
recorder,
internal_events_queue,
manager,
sdk_ready_flag=None,
telemetry_producer=telemetry_producer,
Expand Down Expand Up @@ -992,12 +1004,14 @@ def _build_pluggable_factory(api_key, cfg):
initialization_thread.start()

telemetry_init_producer.record_config(cfg, {}, 0, 0)
internal_events_queue = queue.Queue()

split_factory = SplitFactory(
api_key,
storages,
cfg['labelsEnabled'],
recorder,
internal_events_queue,
manager,
sdk_ready_flag=None,
telemetry_producer=telemetry_producer,
Expand Down Expand Up @@ -1152,11 +1166,14 @@ def _build_localhost_factory(cfg):
telemetry_evaluation_producer,
telemetry_runtime_producer
)
internal_events_queue = queue.Queue()

return SplitFactory(
'localhost',
storages,
False,
recorder,
internal_events_queue,
manager,
ready_event,
telemetry_producer=telemetry_producer,
Expand Down
30 changes: 30 additions & 0 deletions tests/client/test_client.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -66,6 +66,7 @@ def synchronize_config(*_):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -136,6 +137,7 @@ def test_get_treatment_with_config(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -215,6 +217,7 @@ def test_get_treatments(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -296,6 +299,7 @@ def test_get_treatments_by_flag_set(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -376,6 +380,7 @@ def test_get_treatments_by_flag_sets(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -455,6 +460,7 @@ def test_get_treatments_with_config(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -531,6 +537,7 @@ def test_get_treatments_with_config_by_flag_set(self, mocker):
destroyed_property = mocker.PropertyMock()
destroyed_property.return_value = False
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -539,6 +546,7 @@ def test_get_treatments_with_config_by_flag_set(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -620,6 +628,7 @@ def test_get_treatments_with_config_by_flag_sets(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -708,6 +717,7 @@ def synchronize_config(*_):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -773,6 +783,7 @@ def synchronize_config(*_):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -838,6 +849,7 @@ def synchronize_config(*_):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -874,6 +886,7 @@ def test_destroy(self, mocker):
telemetry_storage = InMemoryTelemetryStorage()
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -882,6 +895,7 @@ def test_destroy(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -911,6 +925,7 @@ def test_track(self, mocker):
telemetry_storage = InMemoryTelemetryStorage()
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -919,6 +934,7 @@ def test_track(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -961,6 +977,7 @@ def test_evaluations_before_running_post_fork(self, mocker):

impmanager = mocker.Mock(spec=ImpressionManager)
recorder = StandardRecorder(impmanager, mocker.Mock(), impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -969,6 +986,7 @@ def test_evaluations_before_running_post_fork(self, mocker):
'events': mocker.Mock()},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1047,6 +1065,7 @@ def test_telemetry_not_ready(self, mocker):
'events': mocker.Mock()},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1092,6 +1111,7 @@ def test_telemetry_record_treatment_exception(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
impmanager,
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1193,6 +1213,7 @@ def test_telemetry_method_latency(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
impmanager,
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1255,6 +1276,7 @@ def test_telemetry_track_exception(self, mocker):
telemetry_storage = InMemoryTelemetryStorage()
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -1263,6 +1285,7 @@ def test_telemetry_track_exception(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
impmanager,
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1316,6 +1339,7 @@ def synchronize_config(*_):
'events': event_storage},
mocker.Mock(),
recorder,
events_queue,
mocker.Mock(),
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1412,6 +1436,7 @@ def test_fallback_treatment_eval_exception(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer())
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
internal_events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -1420,6 +1445,7 @@ def test_fallback_treatment_eval_exception(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
internal_events_queue,
impmanager,
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1550,6 +1576,7 @@ def test_fallback_treatment_exception(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer())
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
internal_events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -1558,6 +1585,7 @@ def test_fallback_treatment_exception(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
internal_events_queue,
impmanager,
mocker.Mock(),
telemetry_producer,
Expand Down Expand Up @@ -1618,6 +1646,7 @@ def test_fallback_treatment_not_ready_impressions(self, mocker):
telemetry_producer = TelemetryStorageProducer(telemetry_storage)
impmanager = ImpressionManager(StrategyOptimizedMode(), StrategyNoneMode(), telemetry_producer.get_telemetry_runtime_producer())
recorder = StandardRecorder(impmanager, event_storage, impression_storage, telemetry_producer.get_telemetry_evaluation_producer(), telemetry_producer.get_telemetry_runtime_producer())
internal_events_queue = queue.Queue()
factory = SplitFactory(mocker.Mock(),
{'splits': split_storage,
'segments': segment_storage,
Expand All @@ -1626,6 +1655,7 @@ def test_fallback_treatment_not_ready_impressions(self, mocker):
'events': event_storage},
mocker.Mock(),
recorder,
internal_events_queue,
impmanager,
mocker.Mock(),
telemetry_producer,
Expand Down
Loading