splitio · chillaq · Jan 27, 2026 · Jan 7, 2026 · Jan 8, 2026 · Jan 9, 2026
@@ -3,6 +3,7 @@
 import threading
 from collections import Counter
 from enum import Enum
+import queue 
 
 from splitio.optional.loaders import asyncio
 from splitio.client.client import Client, ClientAsync
@@ -546,9 +547,10 @@ def _build_in_memory_factory(api_key, cfg, sdk_url=None, events_url=None,  # pyl
         'events': EventsAPI(http_client, api_key, sdk_metadata, telemetry_runtime_producer),
         'telemetry': TelemetryAPI(http_client, api_key, sdk_metadata, telemetry_runtime_producer),
     }
-
+
+    events_queue = queue.Queue()
     storages = {
-        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
+        'splits': InMemorySplitStorage(events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),
         'rule_based_segments': InMemoryRuleBasedSegmentStorage(),
         'impressions': InMemoryImpressionStorage(cfg['impressionsQueueSize'], telemetry_runtime_producer),
@@ -1096,8 +1098,9 @@ def _build_localhost_factory(cfg):
     telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
     telemetry_evaluation_producer = telemetry_producer.get_telemetry_evaluation_producer()
 
+    events_queue = queue.Queue()
     storages = {
-        'splits': InMemorySplitStorage(cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
+        'splits': InMemorySplitStorage(events_queue, cfg['flagSetsFilter'] if cfg['flagSetsFilter'] is not None else []),
         'segments': InMemorySegmentStorage(),  # not used, just to avoid possible future errors.
         'rule_based_segments': InMemoryRuleBasedSegmentStorage(),   
         'impressions': LocalhostImpressionsStorage(),

@@ -7,6 +7,9 @@
 from splitio.models.segments import Segment
 from splitio.models.telemetry import HTTPErrors, HTTPLatencies, MethodExceptions, MethodLatencies, LastSynchronization, StreamingEvents, TelemetryConfig, TelemetryCounters, CounterConstants, \
     HTTPErrorsAsync, HTTPLatenciesAsync, MethodExceptionsAsync, MethodLatenciesAsync, LastSynchronizationAsync, StreamingEventsAsync, TelemetryConfigAsync, TelemetryCountersAsync
+from splitio.models.events import SdkInternalEvent
+from splitio.events.events_metadata import EventsMetadata, SdkEventType
+from splitio.models.notification import SdkInternalEventNotification
 from splitio.storage import FlagSetsFilter, SplitStorage, SegmentStorage, ImpressionStorage, EventStorage, TelemetryStorage, RuleBasedSegmentsStorage
 from splitio.optional.loaders import asyncio
 
@@ -479,14 +482,15 @@ def _decrease_traffic_type_count(self, traffic_type_name):
 class InMemorySplitStorage(InMemorySplitStorageBase):
     """InMemory implementation of a feature flag storage."""
 
-    def __init__(self, flag_sets=[]):
+    def __init__(self, internal_event_queue, flag_sets=[]):
         """Constructor."""
         self._lock = threading.RLock()
         self._feature_flags = {}
         self._change_number = -1
         self._traffic_types = Counter()
         self.flag_set = FlagSets(flag_sets)
         self.flag_set_filter = FlagSetsFilter(flag_sets)
+        self._internal_event_queue = internal_event_queue
 
     def clear(self):
         """
@@ -535,6 +539,13 @@ def update(self, to_add, to_delete, new_change_number):
         [self._put(add_feature_flag) for add_feature_flag in to_add]
         [self._remove(delete_feature_flag) for delete_feature_flag in to_delete]
         self._set_change_number(new_change_number)
+        to_notify = []
+        [to_notify.append(feature.name) for feature in to_add]
+        to_notify.extend(to_delete)
+        self._internal_event_queue.put(
+            SdkInternalEventNotification(
+                SdkInternalEvent.FLAGS_UPDATED,
+                EventsMetadata(SdkEventType.FLAG_UPDATE, set(to_notify))))
 
     def _put(self, feature_flag):
         """
@@ -680,6 +691,10 @@ def kill_locally(self, feature_flag_name, default_treatment, change_number):
                 return
             feature_flag.local_kill(default_treatment, change_number)
             self._put(feature_flag)
+            self._internal_event_queue.put(
+                SdkInternalEventNotification(
+                    SdkInternalEvent.FLAG_KILLED_NOTIFICATION,
+                    EventsMetadata(SdkEventType.FLAG_UPDATE, {feature_flag_name})))
 
     def is_flag_set_exist(self, flag_set):
         """

@@ -6,6 +6,7 @@
 import unittest.mock as mock
 import time
 import pytest
+import queue 
 
 from splitio.client.client import Client, _LOGGER as _logger, CONTROL, ClientAsync, EvaluationOptions
 from splitio.client.factory import SplitFactory, Status as FactoryStatus, SplitFactoryAsync
@@ -36,7 +37,8 @@ def test_get_treatment(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -113,7 +115,8 @@ def test_get_treatment_with_config(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -190,7 +193,8 @@ def test_get_treatments(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -270,7 +274,8 @@ def test_get_treatments_by_flag_set(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -349,7 +354,8 @@ def test_get_treatments_by_flag_sets(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -428,7 +434,8 @@ def test_get_treatments_with_config(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -511,7 +518,8 @@ def test_get_treatments_with_config_by_flag_set(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -591,7 +599,8 @@ def test_get_treatments_with_config_by_flag_sets(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -671,7 +680,8 @@ def test_impression_toggle_optimized(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -735,7 +745,8 @@ def test_impression_toggle_debug(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -799,7 +810,8 @@ def test_impression_toggle_none(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
@@ -939,7 +951,8 @@ def test_evaluations_before_running_post_fork(self, mocker):
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
         impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
         impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()        
         split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
@@ -1020,7 +1033,8 @@ def test_telemetry_not_ready(self, mocker):
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()
         impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
         impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()        
         split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
@@ -1053,7 +1067,8 @@ def synchronize_config(*_):
         factory.destroy()
 
     def test_telemetry_record_treatment_exception(self, mocker):
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
         segment_storage = mocker.Mock(spec=SegmentStorage)
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
@@ -1158,7 +1173,8 @@ def test_telemetry_method_latency(self, mocker):
         impression_storage = InMemoryImpressionStorage(10, telemetry_runtime_producer)
         event_storage = mocker.Mock(spec=EventStorage)
         impmanager = ImpressionManager(StrategyDebugMode(), StrategyNoneMode(), telemetry_runtime_producer)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()        
         split_storage.update([from_raw(splits_json['splitChange1_1']['ff']['d'][0])], [], -1)
@@ -1270,7 +1286,8 @@ def test_impressions_properties(self, mocker):
         """Test get_treatment execution paths."""
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        split_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        split_storage = InMemorySplitStorage(events_queue)
         segment_storage = InMemorySegmentStorage()
         rb_segment_storage = InMemoryRuleBasedSegmentStorage()
         telemetry_runtime_producer = telemetry_producer.get_telemetry_runtime_producer()

@@ -1,5 +1,6 @@
 """SDK main manager test module."""
 import pytest
+import queue
 
 from splitio.client.factory import SplitFactory
 from splitio.client.manager import SplitManager, SplitManagerAsync, _LOGGER as _logger
@@ -16,7 +17,8 @@ class SplitManagerTests(object):  # pylint: disable=too-few-public-methods
     def test_manager_calls(self, mocker):
         telemetry_storage = InMemoryTelemetryStorage()
         telemetry_producer = TelemetryStorageProducer(telemetry_storage)
-        storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        storage = InMemorySplitStorage(events_queue)
 
         factory = mocker.Mock(spec=SplitFactory)
         factory._storages = {'split': storage}

@@ -4,6 +4,7 @@
 import os
 import pytest
 import copy
+import queue
 
 from splitio.models.splits import Split, Status, from_raw, Prerequisites
 from splitio.models import segments
@@ -261,7 +262,8 @@ def test_evaluate_treatment_with_rule_based_segment(self, mocker):
 
     def test_evaluate_treatment_with_rbs_in_condition(self):
         e = evaluator.Evaluator(splitters.Splitter())
-        splits_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        splits_storage = InMemorySplitStorage(events_queue)
         rbs_storage = InMemoryRuleBasedSegmentStorage()
         segment_storage = InMemorySegmentStorage()
         evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)
@@ -287,7 +289,8 @@ def test_using_segment_in_excluded(self):
         with open(rbs_segments, 'r') as flo:
             data = json.loads(flo.read())
         e = evaluator.Evaluator(splitters.Splitter())
-        splits_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        splits_storage = InMemorySplitStorage(events_queue)
         rbs_storage = InMemoryRuleBasedSegmentStorage()
         segment_storage = InMemorySegmentStorage()
         evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)
@@ -311,7 +314,8 @@ def test_using_rbs_in_excluded(self):
         with open(rbs_segments, 'r') as flo:
             data = json.loads(flo.read())
         e = evaluator.Evaluator(splitters.Splitter())
-        splits_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        splits_storage = InMemorySplitStorage(events_queue)
         rbs_storage = InMemoryRuleBasedSegmentStorage()
         segment_storage = InMemorySegmentStorage()
         evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)
@@ -334,7 +338,8 @@ def test_prerequisites(self):
         with open(splits_load, 'r') as flo:
             data = json.loads(flo.read())
         e = evaluator.Evaluator(splitters.Splitter())
-        splits_storage = InMemorySplitStorage()
+        events_queue = queue.Queue()
+        splits_storage = InMemorySplitStorage(events_queue)
         rbs_storage = InMemoryRuleBasedSegmentStorage()
         segment_storage = InMemorySegmentStorage()
         evaluation_facctory = EvaluationDataFactory(splits_storage, segment_storage, rbs_storage)
@@ -542,7 +547,8 @@ def test_get_context(self):
         """Test context."""
         mocked_split = Split('some', 12345, False, 'off', 'user', Status.ACTIVE, 12, split_conditions, 1.2, 100, 1234, {}, None, False, [Prerequisites('split2', ['on'])])
         split2 = Split('split2', 12345, False, 'off', 'user', Status.ACTIVE, 12, split_conditions, 1.2, 100, 1234, {}, None, False, [])
-        flag_storage = InMemorySplitStorage([])
+        events_queue = queue.Queue()
+        flag_storage = InMemorySplitStorage(events_queue, [])
         segment_storage = InMemorySegmentStorage()
         rbs_segment_storage = InMemoryRuleBasedSegmentStorage()
         flag_storage.update([mocked_split, split2], [], -1)

@@ -1 +1 @@
-{"ff": {"t": -1, "s": -1, "d": [{"changeNumber": 10, "trafficTypeName": "user", "name": "rbs_feature_flag", "trafficAllocation": 100, "trafficAllocationSeed": 1828377380, "seed": -286617921, "status": "ACTIVE", "killed": false, "defaultTreatment": "off", "algo": 2, "conditions": [{"conditionType": "ROLLOUT", "matcherGroup": {"combiner": "AND", "matchers": [{"keySelector": {"trafficType": "user"}, "matcherType": "IN_RULE_BASED_SEGMENT", "negate": false, "userDefinedSegmentMatcherData": {"segmentName": "sample_rule_based_segment"}}]}, "partitions": [{"treatment": "on", "size": 100}, {"treatment": "off", "size": 0}], "label": "in rule based segment sample_rule_based_segment"}, {"conditionType": "ROLLOUT", "matcherGroup": {"combiner": "AND", "matchers": [{"keySelector": {"trafficType": "user"}, "matcherType": "ALL_KEYS", "negate": false}]}, "partitions": [{"treatment": "on", "size": 0}, {"treatment": "off", "size": 100}], "label": "default rule"}], "configurations": {}, "sets": [], "impressionsDisabled": false}]}, "rbs": {"t": 1675259356568, "s": -1, "d": [{"changeNumber": 5, "name": "sample_rule_based_segment", "status": "ACTIVE", "trafficTypeName": "user", "excluded": {"keys": ["mauro@split.io", "gaston@split.io"], "segments": []}, "conditions": [{"matcherGroup": {"combiner": "AND", "matchers": [{"keySelector": {"trafficType": "user", "attribute": "email"}, "matcherType": "ENDS_WITH", "negate": false, "whitelistMatcherData": {"whitelist": ["@split.io"]}}]}}]}]}}
+{"ff": {"t": -1, "s": -1, "d": [{"name": "SPLIT_1", "status": "ACTIVE", "killed": false, "defaultTreatment": "off", "configurations": {}, "conditions": []}]}, "rbs": {"t": -1, "s": -1, "d": [{"changeNumber": 12, "name": "some_segment", "status": "ACTIVE", "trafficTypeName": "user", "excluded": {"keys": [], "segments": []}, "conditions": []}]}}
Original file line number	Diff line number	Diff line change
		@@ -1 +1 @@
		{"ff": {"t": -1, "s": -1, "d": [{"changeNumber": 10, "trafficTypeName": "user", "name": "rbs_feature_flag", "trafficAllocation": 100, "trafficAllocationSeed": 1828377380, "seed": -286617921, "status": "ACTIVE", "killed": false, "defaultTreatment": "off", "algo": 2, "conditions": [{"conditionType": "ROLLOUT", "matcherGroup": {"combiner": "AND", "matchers": [{"keySelector": {"trafficType": "user"}, "matcherType": "IN_RULE_BASED_SEGMENT", "negate": false, "userDefinedSegmentMatcherData": {"segmentName": "sample_rule_based_segment"}}]}, "partitions": [{"treatment": "on", "size": 100}, {"treatment": "off", "size": 0}], "label": "in rule based segment sample_rule_based_segment"}, {"conditionType": "ROLLOUT", "matcherGroup": {"combiner": "AND", "matchers": [{"keySelector": {"trafficType": "user"}, "matcherType": "ALL_KEYS", "negate": false}]}, "partitions": [{"treatment": "on", "size": 0}, {"treatment": "off", "size": 100}], "label": "default rule"}], "configurations": {}, "sets": [], "impressionsDisabled": false}]}, "rbs": {"t": 1675259356568, "s": -1, "d": [{"changeNumber": 5, "name": "sample_rule_based_segment", "status": "ACTIVE", "trafficTypeName": "user", "excluded": {"keys": ["mauro@split.io", "gaston@split.io"], "segments": []}, "conditions": [{"matcherGroup": {"combiner": "AND", "matchers": [{"keySelector": {"trafficType": "user", "attribute": "email"}, "matcherType": "ENDS_WITH", "negate": false, "whitelistMatcherData": {"whitelist": ["@split.io"]}}]}}]}]}}
		{"ff": {"t": -1, "s": -1, "d": [{"name": "SPLIT_1", "status": "ACTIVE", "killed": false, "defaultTreatment": "off", "configurations": {}, "conditions": []}]}, "rbs": {"t": -1, "s": -1, "d": [{"changeNumber": 12, "name": "some_segment", "status": "ACTIVE", "trafficTypeName": "user", "excluded": {"keys": [], "segments": []}, "conditions": []}]}}