Merge pull request #323 from splitio/FME-12320-events-task

added InternalEventsTask to use a queue for firing internal events

Author: sanzmauro

CHANGES.txt

@@ -1,5 +1,11 @@
 CHANGES
 
+7.13.0 (Jan 30, 2026)
+- Fixed impressions properties format in redis mode.
+- Added the ability to listen to different events triggered by the SDK. Read more in our docs.
+   - SDK_UPDATE notify when a flag or user segment has changed
+   - SDK_READY notify when the SDK is ready to evaluate
+
 7.12.0 (Sep 30, 2025)
 - Added new configuration for Fallback Treatments, which allows setting a treatment value and optional config to be returned in place of "control", either globally or by flag. Read more in our docs.
 - Added a maximum size payload when posting unique keys telemetry in batches

Splitio.Redis/Services/Cache/Classes/RedisImpressionsCache.cs

@@ -78,12 +78,13 @@ public async Task RecordImpressionsCountAsync(Dictionary<string, int> impression
             }
         }
 
-        private RedisValue[] GetImpressions(IList<KeyImpression> items)
+        // public for tests
+        public RedisValue[] GetImpressions(IList<KeyImpression> items)
         {
-            var impressions = items.Select(item => JsonConvertWrapper.SerializeObject(new
+            var impressions = items.Select(item => JsonConvertWrapper.SerializeObjectIgnoreNullValue(new
             {
                 m = new { s = SdkVersion, i = MachineIp, n = MachineName },
-                i = new { k = item.keyName, b = item.bucketingKey, f = item.feature, t = item.treatment, r = item.label, c = item.changeNumber, m = item.time, pt = item.previousTime }
+                i = new { k = item.keyName, b = item.bucketingKey, f = item.feature, t = item.treatment, r = item.label, c = item.changeNumber, m = item.time, pt = item.previousTime, properties = item.properties }
             }));
 
             return impressions

Splitio.Redis/Splitio.Redis.csproj

@@ -7,7 +7,7 @@
     <GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute>
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
-    <Version>7.12.0</Version>
+    <Version>7.13.0-rc1</Version>
     <SignAssembly>true</SignAssembly>
     <AssemblyOriginatorKeyFile>SplitioRedis.snk</AssemblyOriginatorKeyFile>
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>

src/Splitio/Domain/EventMetadata.cs

@@ -1,13 +1,11 @@
 using System.Collections.Generic;
-using Splitio.Services.InputValidation.Classes;
-using System.Linq;
 
 namespace Splitio.Domain
 {
     public class EventMetadata
     {
-        private List<string> _names;
-        private SdkEventType _type;
+        private readonly List<string> _names;
+        private readonly SdkEventType _type;
 
         public EventMetadata(SdkEventType type, List<string> names) 
         {

src/Splitio/Domain/EventsManagerConfig.cs

@@ -36,31 +36,19 @@ public EventsManagerConfig()
                         SdkInternalEvent.FlagKilledNotification,
                         SdkInternalEvent.SegmentsUpdated
                     }
-                },
-                { SdkEvent.SdkReadyTimeout, new HashSet<SdkInternalEvent>
-                    {
-                        SdkInternalEvent.SdkTimedOut
-                    }
                 }
             };
 
-            SuppressedBy = new Dictionary<SdkEvent, HashSet<SdkEvent>>
-            {
-                { SdkEvent.SdkReadyTimeout, new HashSet<SdkEvent>
-                    { SdkEvent.SdkReady }
-
-                }
-            };
+            SuppressedBy = new Dictionary<SdkEvent, HashSet<SdkEvent>>();
 
             ExecutionLimits = new Dictionary<SdkEvent, int>
             {
-                { SdkEvent.SdkReadyTimeout, -1 },
                 { SdkEvent.SdkReady, 1 },
                 { SdkEvent.SdkUpdate, -1 }
             };
 
             HashSet<SdkEvent> sortedEvents = new HashSet<SdkEvent>();
-            foreach (SdkEvent sdkEvent in new List<SdkEvent> { SdkEvent.SdkReady, SdkEvent.SdkReadyTimeout, SdkEvent.SdkUpdate }) 
+            foreach (SdkEvent sdkEvent in new List<SdkEvent> { SdkEvent.SdkReady, SdkEvent.SdkUpdate }) 
             {
                 sortedEvents = DFSRecursive(sdkEvent, sortedEvents);
             }

src/Splitio/Domain/SdkEvent.cs

@@ -4,7 +4,6 @@ namespace Splitio.Domain
     public enum SdkEvent
     {
         SdkUpdate,
-        SdkReadyTimeout,
         SdkReady
     }
 }

src/Splitio/Domain/SdkEventNotification.cs

@@ -0,0 +1,16 @@
+using Splitio.Domain;
+
+namespace Splitio.Services.EventSource.Workers
+{
+    public class SdkEventNotification
+    {
+        public SdkInternalEvent SdkInternalEvent { get; set; }
+        public EventMetadata EventMetadata { get; set; }
+
+        public SdkEventNotification(SdkInternalEvent sdkInternalEvent, EventMetadata eventMetadata)
+        {
+            SdkInternalEvent = sdkInternalEvent;
+            EventMetadata = eventMetadata;
+        }
+    }
+}

src/Splitio/Domain/SdkInternalEvent.cs

@@ -8,7 +8,6 @@ public enum SdkInternalEvent
         RuleBasedSegmentsUpdated,
         SegmentsUpdated,
         LargeSegmentsUpdated,
-        SdkTimedOut,
         SdkReady
     }
 }

src/Splitio/Services/Cache/Classes/InMemoryReadinessGatesCache.cs

@@ -1,19 +1,20 @@
 using Splitio.Domain;
 using Splitio.Services.Cache.Interfaces;
-using Splitio.Services.Common;
+using Splitio.Services.Tasks;
 using System.Threading;
+using System.Threading.Tasks;
 
 namespace Splitio.Services.Client.Classes
 {
     public class InMemoryReadinessGatesCache : IStatusManager
     {
         private readonly CountdownEvent _sdkReady = new CountdownEvent(1);
         private readonly CountdownEvent _sdkDestroyed = new CountdownEvent(1);
-        private readonly IEventsManager<SdkEvent, SdkInternalEvent, EventMetadata> _eventsManager;
+        private readonly IInternalEventsTask _internalEventsTask;
 
-        public InMemoryReadinessGatesCache(IEventsManager<SdkEvent, SdkInternalEvent, EventMetadata> eventsManager) 
+        public InMemoryReadinessGatesCache(IInternalEventsTask internalEventsTask)
         {
-            _eventsManager = eventsManager;
+            _internalEventsTask = internalEventsTask;
         }
 
         public bool IsReady()
@@ -26,10 +27,10 @@ public bool WaitUntilReady(int milliseconds)
             return _sdkReady.Wait(milliseconds);
         }
 
-        public void SetReady()
+        public async Task SetReadyAsync()
         {
             _sdkReady.Signal();
-            _eventsManager.NotifyInternalEvent(SdkInternalEvent.SdkReady, null);
+            await _internalEventsTask.AddToQueue(SdkInternalEvent.SdkReady, null);
         }
 
         public void SetDestroy()

src/Splitio/Services/Cache/Classes/InMemoryRuleBasedSegmentCache.cs

@@ -1,6 +1,8 @@
 using Splitio.Domain;
 using Splitio.Services.Cache.Interfaces;
-using Splitio.Services.Common;
+using Splitio.Services.Logger;
+using Splitio.Services.Shared.Classes;
+using Splitio.Services.Tasks;
 using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.Threading.Tasks;
@@ -11,15 +13,16 @@ public class InMemoryRuleBasedSegmentCache : IRuleBasedSegmentCache
     {
         private readonly ConcurrentDictionary<string, RuleBasedSegment> _cache;
         private long _changeNumber;
-        private readonly IEventsManager<SdkEvent, SdkInternalEvent, EventMetadata> _eventsManager;
+        private readonly IInternalEventsTask _internalEventsTask;
+        private readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(InMemoryRuleBasedSegmentCache));
 
         public InMemoryRuleBasedSegmentCache(ConcurrentDictionary<string, RuleBasedSegment> cache,
-            IEventsManager<SdkEvent, SdkInternalEvent, EventMetadata> eventsManger,
+            IInternalEventsTask internalEventsTask,
             long changeNumber = -1)
         {
             _cache = cache;
             _changeNumber = changeNumber;
-            _eventsManager = eventsManger;
+            _internalEventsTask = internalEventsTask;
         }
 
         #region Sync Methods
@@ -60,8 +63,12 @@ public void Update(List<RuleBasedSegment> toAdd, List<string> toRemove, long til
             }
 
             SetChangeNumber(till);
-            _eventsManager.NotifyInternalEvent(SdkInternalEvent.RuleBasedSegmentsUpdated,
-                new EventMetadata(SdkEventType.SegmentsUpdate, new List<string>()));
+            Task task = new Task(() =>
+            {
+                _internalEventsTask.AddToQueue(SdkInternalEvent.RuleBasedSegmentsUpdated,
+                new EventMetadata(SdkEventType.SegmentsUpdate, new List<string>())).ContinueWith(OnAddToQueueFailed, TaskContinuationOptions.OnlyOnFaulted);
+            });
+            task.Start();
         }
 
         public void SetChangeNumber(long changeNumber)
@@ -81,5 +88,10 @@ public Task<RuleBasedSegment> GetAsync(string name)
             return Task.FromResult(Get(name));
         }
         #endregion
+
+        public void OnAddToQueueFailed(Task task)
+        {
+            _log.Error($"Failed to add internal event to queue: {task.Exception.Message}");
+        }
     }
 }