[7.8.3] Development into main

CHANGES.txt

@@ -1,5 +1,10 @@
 CHANGES
 
+7.8.3 (Apr 17, 2024)
+- Fixed Sync over Async implementation.
+- Added the logic to handle correctly when the SDK receives an unsupported Matcher type.
+- Fixed NotificationManagerKeeper to start with publishers in both regions by default. 
+
 7.8.2 (Mar 25, 2024)
 - Fixed streaming notification parser.
 

Splitio.Redis/Services/Client/Classes/RedisClient.cs

@@ -5,7 +5,6 @@
 using Splitio.Redis.Services.Domain;
 using Splitio.Redis.Services.Events.Classes;
 using Splitio.Redis.Services.Impressions.Classes;
-using Splitio.Redis.Services.Parsing.Classes;
 using Splitio.Redis.Services.Shared;
 using Splitio.Redis.Telemetry.Storages;
 using Splitio.Services.Cache.Interfaces;
@@ -15,6 +14,7 @@
 using Splitio.Services.Impressions.Classes;
 using Splitio.Services.Impressions.Interfaces;
 using Splitio.Services.InputValidation.Classes;
+using Splitio.Services.Parsing;
 using Splitio.Services.Shared.Classes;
 
 namespace Splitio.Redis.Services.Client.Classes
@@ -83,7 +83,7 @@ private void BuildRedisCache()
             BuildTelemetryStorage();
 
             _segmentCacheConsumer = new RedisSegmentCache(_redisAdapter, _config.RedisUserPrefix);
-            _splitParser = new RedisSplitParser(_segmentCacheConsumer);
+            _splitParser = new SplitParser(_segmentCacheConsumer);
             _featureFlagCacheConsumer = new RedisSplitCache(_redisAdapter, _splitParser, _config.RedisUserPrefix);
             _blockUntilReadyService = new RedisBlockUntilReadyService(_redisAdapter);
             _trafficTypeValidator = new TrafficTypeValidator(_featureFlagCacheConsumer, _blockUntilReadyService);

Splitio.Redis/Splitio.Redis.csproj

@@ -7,7 +7,7 @@
     <GenerateAssemblyConfigurationAttribute>false</GenerateAssemblyConfigurationAttribute>
     <GenerateAssemblyCompanyAttribute>false</GenerateAssemblyCompanyAttribute>
     <GenerateAssemblyProductAttribute>false</GenerateAssemblyProductAttribute>
-    <Version>7.8.2</Version>
+    <Version>7.8.3</Version>
     <SignAssembly>true</SignAssembly>
     <AssemblyOriginatorKeyFile>SplitioRedis.snk</AssemblyOriginatorKeyFile>
     <PackageLicenseExpression>Apache-2.0</PackageLicenseExpression>
@@ -36,4 +36,8 @@
     <ProjectReference Include="..\src\Splitio\Splitio.csproj" />
   </ItemGroup>
 
+  <ItemGroup>
+    <Folder Include="Services\Parsing\Classes\" />
+  </ItemGroup>
+
 </Project>

appveyor.yml

@@ -4,7 +4,8 @@ image:
 nuget:
   account_feed: true
 
-version: 7.8.2
+version: 7.8.3
+
 dotnet_csproj:
   patch: true
   file: '**\*.csproj'

src/Splitio/Domain/Labels.cs

@@ -8,5 +8,6 @@ public class Labels
         public static string Exception => "exception";
         public static string TrafficAllocationFailed => "not in split";
         public static string ClientNotReady => "not ready";
+        public static string UnsupportedMatcherType => "unsupported matcher type";
     }
 }

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

@@ -10,7 +10,7 @@ namespace Splitio.Services.Cache.Classes
 {
     public class InMemorySegmentCache : ISegmentCache
     {
-        private static readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(InMemorySegmentCache));
+        private readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(InMemorySegmentCache));
 
         private readonly ConcurrentDictionary<string, Segment> _segments;
 

src/Splitio/Services/Client/Classes/SelfRefreshingClient.cs

@@ -106,12 +106,12 @@ private void BuildTelemetryStorage()
         private void BuildSplitFetcher()
         {
             var segmentChangeFetcher = new ApiSegmentChangeFetcher(_segmentSdkApiClient);
-            var segmentTaskQueue = new BlockingCollection<SelfRefreshingSegment>(new ConcurrentQueue<SelfRefreshingSegment>());
             var segmentRefreshRate = _config.RandomizeRefreshRates ? Random(_config.SegmentRefreshRate) : _config.SegmentRefreshRate;
-            var workerTask = _tasksManager.NewPeriodicTask(Enums.Task.SegmentsWorkerFetcher, 0);
-            var segmentFetcherWorkerPool = new SegmentTaskWorker(_config.NumberOfParalellSegmentTasks, segmentTaskQueue, _statusManager, workerTask, _tasksManager);
+            var segmentsQueue = new SplitQueue<SelfRefreshingSegment>();
+            var segmentFetcherWorkerPool = new SegmentTaskWorker(_config.NumberOfParalellSegmentTasks, segmentsQueue);
+            segmentsQueue.AddObserver(segmentFetcherWorkerPool);
             var segmentsFetcherTask = _tasksManager.NewPeriodicTask(Enums.Task.SegmentsFetcher, segmentRefreshRate * 1000);
-            _selfRefreshingSegmentFetcher = new SelfRefreshingSegmentFetcher(segmentChangeFetcher, _segmentCache, segmentTaskQueue, segmentsFetcherTask, segmentFetcherWorkerPool, _statusManager);
+            _selfRefreshingSegmentFetcher = new SelfRefreshingSegmentFetcher(segmentChangeFetcher, _segmentCache, segmentsQueue, segmentsFetcherTask, _statusManager);
 
             var splitChangeFetcher = new ApiSplitChangeFetcher(_splitSdkApiClient);
             _splitParser = new InMemorySplitParser((SelfRefreshingSegmentFetcher)_selfRefreshingSegmentFetcher, _segmentCache);
@@ -223,11 +223,8 @@ private void BuildSyncManager()
                 var synchronizer = new Synchronizer(_splitFetcher, _selfRefreshingSegmentFetcher, _impressionsLog, _eventsLog, _impressionsCounter, _statusManager, _telemetrySyncTask, _featureFlagCache, backOffFeatureFlags, backOffSegments, _config.OnDemandFetchMaxRetries, _config.OnDemandFetchRetryDelayMs, _segmentCache, _uniqueKeysTracker);
 
                 // Workers
-                var queue = new BlockingCollection<SplitChangeNotification>(new ConcurrentQueue<SplitChangeNotification>());
-                var featureFlagsWorkerTask = _tasksManager.NewPeriodicTask(Enums.Task.FeatureFlagsWorker, 0);
-                var splitsWorker = new SplitsWorker(synchronizer, _featureFlagCache, queue, _telemetryRuntimeProducer, _selfRefreshingSegmentFetcher, featureFlagsWorkerTask, _featureFlagSyncService);
-                var segmentsWorkerTask = _tasksManager.NewPeriodicTask(Enums.Task.SegmentsWorker, 0);
-                var segmentsWorker = new SegmentsWorker(synchronizer, segmentsWorkerTask);
+                var splitsWorker = new SplitsWorker(synchronizer, _featureFlagCache, _telemetryRuntimeProducer, _selfRefreshingSegmentFetcher, _featureFlagSyncService);
+                var segmentsWorker = new SegmentsWorker(synchronizer);
 
                 // NotificationProcessor
                 var notificationProcessor = new NotificationProcessor(splitsWorker, segmentsWorker);
@@ -236,7 +233,7 @@ private void BuildSyncManager()
                 var notificationParser = new NotificationParser();
 
                 // SSEClient Status actions queue
-                var streamingStatusQueue = new BlockingCollection<StreamingStatus>(new ConcurrentQueue<StreamingStatus>());
+                var streamingStatusQueue = new SplitQueue<StreamingStatus>();
 
                 // NotificationManagerKeeper
                 var notificationManagerKeeper = new NotificationManagerKeeper(_telemetryRuntimeProducer, streamingStatusQueue);
@@ -263,8 +260,7 @@ private void BuildSyncManager()
                 // SyncManager
                 var streamingbackoff = new BackOff(_config.StreamingReconnectBackoffBase, attempt: 1);
                 var startupTask = _tasksManager.NewOnTimeTask(Enums.Task.SDKInitialization);
-                var streamingStatusTask = _tasksManager.NewPeriodicTask(Enums.Task.OnStreamingStatusTask, 0);
-                _syncManager = new SyncManager(_config.StreamingEnabled, synchronizer, pushManager, sseHandler, _telemetryRuntimeProducer, _statusManager, _tasksManager, _telemetrySyncTask, streamingStatusQueue, streamingbackoff, startupTask, streamingStatusTask);
+                _syncManager = new SyncManager(_config.StreamingEnabled, synchronizer, pushManager, sseHandler, _telemetryRuntimeProducer, _statusManager, _tasksManager, _telemetrySyncTask, streamingbackoff, streamingStatusQueue, startupTask);
             }
             catch (Exception ex)
             {

src/Splitio/Services/Client/Classes/SplitManager.cs

@@ -14,7 +14,7 @@ namespace Splitio.Services.Client.Classes
 {
     public class SplitManager : ISplitManager
     {
-        private static readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(SplitManager));
+        private readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(SplitManager));
 
         private readonly IFeatureFlagCacheConsumer _featureFlagCacheConsumer;
         private readonly ISplitNameValidator _splitNameValidator;
@@ -117,7 +117,7 @@ private bool IsFeatureFlagNameValid(Enums.API method, string featureName, out st
             return result.Success;
         }
 
-        private static SplitView GetFeatureFlagView(ParsedSplit featureFlag, string featureName)
+        private SplitView GetFeatureFlagView(ParsedSplit featureFlag, string featureName)
         {
             if (featureFlag == null)
             {

src/Splitio/Services/Common/AuthApiClient.cs

@@ -14,7 +14,7 @@ namespace Splitio.Services.Common
 {
     public class AuthApiClient : IAuthApiClient
     {
-        private static readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(AuthApiClient));
+        private readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(AuthApiClient));
 
         private readonly ISplitioHttpClient _splitioHttpClient;
         private readonly ITelemetryRuntimeProducer _telemetryRuntimeProducer;

src/Splitio/Services/Common/PushManager.cs

@@ -56,19 +56,19 @@ public async Task StartAsync()
 
                 if (!response.PushEnabled.Value && !response.Retry.Value)
                 {
-                    _notificationManagerKeeper.HandleSseStatus(SSEClientStatusMessage.FORCED_STOP);
+                    await _notificationManagerKeeper.HandleSseStatus(SSEClientStatusMessage.FORCED_STOP);
                     return;
                 }
 
-                if (response.PushEnabled.Value && _sseHandler.Start(response.Token, response.Channels))
+                if (response.PushEnabled.Value)
                 {
+                    _sseHandler.Start(response.Token, response.Channels);
                     _intervalToken = response.Expiration.Value;
-                    _telemetryRuntimeProducer.RecordStreamingEvent(new StreamingEvent(EventTypeEnum.TokenRefresh, CalcularteNextTokenExpiration(_intervalToken)));
                     return;
                 }
 
                 if (!_statusManager.IsDestroyed() && response.Retry.Value)
-                    _notificationManagerKeeper.HandleSseStatus(SSEClientStatusMessage.RETRYABLE_ERROR);
+                    await _notificationManagerKeeper.HandleSseStatus(SSEClientStatusMessage.RETRYABLE_ERROR);
             }
             catch (Exception ex)
             {
@@ -96,7 +96,8 @@ public async Task ScheduleConnectionResetAsync()
                 _log.Debug("ScheduleConnectionReset task is running. Stoping and creating a new one.");
                 await _refreshTokenTask.StopAsync();
             }
-
+
+            _telemetryRuntimeProducer.RecordStreamingEvent(new StreamingEvent(EventTypeEnum.TokenRefresh, CalcularteNextTokenExpiration(_intervalToken)));
             var intervalTime = Convert.ToInt32(_intervalToken) * 1000;
             _log.Debug($"ScheduleNextTokenRefresh interval time : {intervalTime} milliseconds.");
 

src/Splitio/Services/Common/SplitioHttpClient.cs

@@ -15,7 +15,7 @@ namespace Splitio.Services.Common
 {
     public class SplitioHttpClient : ISplitioHttpClient
     {
-        private static readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(SplitioHttpClient));
+        private readonly ISplitLogger _log = WrapperAdapter.Instance().GetLogger(typeof(SplitioHttpClient));
 
         private readonly HttpClient _httpClient;