Process RBS streaming notification (#748)

Author: gthea

src/androidTest/java/fake/SynchronizerSpyImpl.java

@@ -46,6 +46,11 @@ public void synchronizeSplits(long since) {
         mSynchronizer.synchronizeSplits();
     }
 
+    @Override
+    public void synchronizeRuleBasedSegments(long changeNumber) {
+        mSynchronizer.synchronizeRuleBasedSegments(changeNumber);
+    }
+
     @Override
     public void synchronizeSplits() {
         mSynchronizer.synchronizeSplits();

src/main/java/io/split/android/client/SplitFactoryHelper.java

@@ -45,6 +45,7 @@
 import io.split.android.client.service.sseclient.ReconnectBackoffCounter;
 import io.split.android.client.service.sseclient.SseJwtParser;
 import io.split.android.client.service.sseclient.feedbackchannel.PushManagerEventBroadcaster;
+import io.split.android.client.service.sseclient.notifications.InstantUpdateChangeNotification;
 import io.split.android.client.service.sseclient.notifications.MySegmentsV2PayloadDecoder;
 import io.split.android.client.service.sseclient.notifications.NotificationParser;
 import io.split.android.client.service.sseclient.notifications.NotificationProcessor;
@@ -86,6 +87,7 @@
 import io.split.android.client.storage.events.PersistentEventsStorage;
 import io.split.android.client.storage.general.GeneralInfoStorage;
 import io.split.android.client.storage.impressions.PersistentImpressionsStorage;
+import io.split.android.client.storage.rbs.RuleBasedSegmentStorage;
 import io.split.android.client.storage.splits.SplitsStorage;
 import io.split.android.client.telemetry.TelemetrySynchronizer;
 import io.split.android.client.telemetry.TelemetrySynchronizerImpl;
@@ -353,7 +355,7 @@ public StreamingComponents buildStreamingComponents(@NonNull SplitTaskExecutor s
             return new StreamingComponents();
         }
 
-        BlockingQueue<SplitsChangeNotification> splitsUpdateNotificationQueue = new LinkedBlockingDeque<>();
+        BlockingQueue<InstantUpdateChangeNotification> splitsUpdateNotificationQueue = new LinkedBlockingDeque<>();
         NotificationParser notificationParser = new NotificationParser();
 
         NotificationProcessor notificationProcessor = new NotificationProcessor(splitTaskExecutor, splitTaskFactory,
@@ -419,13 +421,15 @@ SplitUpdatesWorker getSplitUpdatesWorker(SplitClientConfig config,
                                              SplitTaskExecutor splitTaskExecutor,
                                              SplitTaskFactory splitTaskFactory,
                                              Synchronizer mSynchronizer,
-                                             BlockingQueue<SplitsChangeNotification> splitsUpdateNotificationQueue,
+                                             BlockingQueue<InstantUpdateChangeNotification> splitsUpdateNotificationQueue,
                                              SplitsStorage splitsStorage,
+                                             RuleBasedSegmentStorage ruleBasedSegmentStorage,
                                              CompressionUtilProvider compressionProvider) {
         if (config.syncEnabled()) {
             return new SplitUpdatesWorker(mSynchronizer,
                     splitsUpdateNotificationQueue,
                     splitsStorage,
+                    ruleBasedSegmentStorage,
                     compressionProvider,
                     splitTaskExecutor,
                     splitTaskFactory);

src/main/java/io/split/android/client/SplitFactoryImpl.java

@@ -241,6 +241,7 @@ private SplitFactoryImpl(@NonNull String apiToken, @NonNull Key key, @NonNull Sp
                         mSynchronizer,
                         streamingComponents.getSplitsUpdateNotificationQueue(),
                         mStorageContainer.getSplitsStorage(),
+                        mStorageContainer.getRuleBasedSegmentStorage(),
                         compressionProvider),
                 streamingComponents.getSyncGuardian());
 

src/main/java/io/split/android/client/events/SplitInternalEvent.java

@@ -16,4 +16,5 @@ public enum SplitInternalEvent {
     ATTRIBUTES_LOADED_FROM_STORAGE,
     ENCRYPTION_MIGRATION_DONE,
     MY_LARGE_SEGMENTS_UPDATED,
+    RULE_BASED_SEGMENTS_UPDATED,
 }

src/main/java/io/split/android/client/service/executor/SplitTaskFactory.java

@@ -1,12 +1,14 @@
 package io.split.android.client.service.executor;
 
+import io.split.android.client.dtos.RuleBasedSegment;
 import io.split.android.client.dtos.Split;
 import io.split.android.client.service.CleanUpDatabaseTask;
 import io.split.android.client.service.events.EventsRecorderTask;
 import io.split.android.client.service.impressions.ImpressionsTaskFactory;
 import io.split.android.client.service.rules.LoadRuleBasedSegmentsTask;
 import io.split.android.client.service.splits.FilterSplitsInCacheTask;
 import io.split.android.client.service.splits.LoadSplitsTask;
+import io.split.android.client.service.splits.RuleBasedSegmentInPlaceUpdateTask;
 import io.split.android.client.service.splits.SplitInPlaceUpdateTask;
 import io.split.android.client.service.splits.SplitKillTask;
 import io.split.android.client.service.splits.SplitsSyncTask;
@@ -37,4 +39,6 @@ public interface SplitTaskFactory extends TelemetryTaskFactory, ImpressionsTaskF
     CleanUpDatabaseTask createCleanUpDatabaseTask(long maxTimestamp);
 
     EncryptionMigrationTask createEncryptionMigrationTask(String sdkKey, SplitRoomDatabase splitRoomDatabase, boolean encryptionEnabled, SplitCipher splitCipher);
+
+    RuleBasedSegmentInPlaceUpdateTask createRuleBasedSegmentUpdateTask(RuleBasedSegment ruleBasedSegment, long changeNumber);
 }

src/main/java/io/split/android/client/service/executor/SplitTaskFactoryImpl.java

@@ -16,6 +16,7 @@
 import io.split.android.client.SplitClientConfig;
 import io.split.android.client.SplitFilter;
 import io.split.android.client.TestingConfig;
+import io.split.android.client.dtos.RuleBasedSegment;
 import io.split.android.client.dtos.Split;
 import io.split.android.client.events.ISplitEventsManager;
 import io.split.android.client.service.CleanUpDatabaseTask;
@@ -32,8 +33,10 @@
 import io.split.android.client.service.impressions.unique.UniqueKeysRecorderTask;
 import io.split.android.client.service.impressions.unique.UniqueKeysRecorderTaskConfig;
 import io.split.android.client.service.rules.LoadRuleBasedSegmentsTask;
+import io.split.android.client.service.rules.RuleBasedSegmentChangeProcessor;
 import io.split.android.client.service.splits.FilterSplitsInCacheTask;
 import io.split.android.client.service.splits.LoadSplitsTask;
+import io.split.android.client.service.splits.RuleBasedSegmentInPlaceUpdateTask;
 import io.split.android.client.service.splits.SplitChangeProcessor;
 import io.split.android.client.service.splits.SplitInPlaceUpdateTask;
 import io.split.android.client.service.splits.SplitKillTask;
@@ -64,6 +67,7 @@ public class SplitTaskFactoryImpl implements SplitTaskFactory {
     private final ISplitEventsManager mEventsManager;
     private final TelemetryTaskFactory mTelemetryTaskFactory;
     private final SplitChangeProcessor mSplitChangeProcessor;
+    private final RuleBasedSegmentChangeProcessor mRuleBasedSegmentChangeProcessor;
     private final TelemetryRuntimeProducer mTelemetryRuntimeProducer;
     private final List<SplitFilter> mFilters;
 
@@ -85,6 +89,7 @@ public SplitTaskFactoryImpl(@NonNull SplitClientConfig splitClientConfig,
         mFlagsSpecFromConfig = flagsSpecFromConfig;
         mEventsManager = eventsManager;
         mSplitChangeProcessor = new SplitChangeProcessor(filters, flagSetsFilter);
+        mRuleBasedSegmentChangeProcessor = new RuleBasedSegmentChangeProcessor();
         RuleBasedSegmentStorageProducer ruleBasedSegmentStorageProducer = mSplitsStorageContainer.getRuleBasedSegmentStorage();
 
         TelemetryStorage telemetryStorage = mSplitsStorageContainer.getTelemetryStorage();
@@ -93,6 +98,7 @@ public SplitTaskFactoryImpl(@NonNull SplitClientConfig splitClientConfig,
             mSplitsSyncHelper = new SplitsSyncHelper(mSplitApiFacade.getSplitFetcher(),
                     mSplitsStorageContainer.getSplitsStorage(),
                     mSplitChangeProcessor,
+                    mRuleBasedSegmentChangeProcessor,
                     ruleBasedSegmentStorageProducer,
                     mTelemetryRuntimeProducer,
                     new ReconnectBackoffCounter(1, testingConfig.getCdnBackoffTime()),
@@ -101,6 +107,7 @@ public SplitTaskFactoryImpl(@NonNull SplitClientConfig splitClientConfig,
             mSplitsSyncHelper = new SplitsSyncHelper(mSplitApiFacade.getSplitFetcher(),
                     mSplitsStorageContainer.getSplitsStorage(),
                     mSplitChangeProcessor,
+                    mRuleBasedSegmentChangeProcessor,
                     ruleBasedSegmentStorageProducer,
                     mTelemetryRuntimeProducer,
                     flagsSpecFromConfig);
@@ -222,6 +229,11 @@ public EncryptionMigrationTask createEncryptionMigrationTask(String sdkKey, Spli
         return new EncryptionMigrationTask(sdkKey, splitRoomDatabase, encryptionEnabled, splitCipher);
     }
 
+    @Override
+    public RuleBasedSegmentInPlaceUpdateTask createRuleBasedSegmentUpdateTask(RuleBasedSegment ruleBasedSegment, long changeNumber) {
+        return new RuleBasedSegmentInPlaceUpdateTask(mSplitsStorageContainer.getRuleBasedSegmentStorage(), mRuleBasedSegmentChangeProcessor, mEventsManager, ruleBasedSegment, changeNumber);
+    }
+
     @NonNull
     private TelemetryTaskFactory initializeTelemetryTaskFactory(@NonNull SplitClientConfig splitClientConfig, @Nullable Map<SplitFilter.Type, SplitFilter> filters, TelemetryStorage telemetryStorage) {
         final TelemetryTaskFactory mTelemetryTaskFactory;

src/main/java/io/split/android/client/service/executor/SplitTaskType.java

@@ -9,4 +9,5 @@ public enum SplitTaskType {
     TELEMETRY_CONFIG_TASK, TELEMETRY_STATS_TASK,
     SAVE_UNIQUE_KEYS_TASK, UNIQUE_KEYS_RECORDER_TASK,
     MY_LARGE_SEGMENTS_UPDATE, LOAD_LOCAL_RULE_BASED_SEGMENTS,
+    RULE_BASED_SEGMENT_SYNC,
 }

src/main/java/io/split/android/client/service/rules/ProcessedRuleBasedSegmentChange.java

@@ -0,0 +1,38 @@
+package io.split.android.client.service.rules;
+
+import java.util.Set;
+
+import io.split.android.client.dtos.RuleBasedSegment;
+
+public class ProcessedRuleBasedSegmentChange {
+    private final Set<RuleBasedSegment> mActive;
+    private final Set<RuleBasedSegment> mArchived;
+    private final long mChangeNumber;
+    private final long mUpdateTimestamp;
+
+    public ProcessedRuleBasedSegmentChange(Set<RuleBasedSegment> active,
+                                           Set<RuleBasedSegment> archived,
+                                           long changeNumber,
+                                           long updateTimestamp) {
+        mActive = active;
+        mArchived = archived;
+        mChangeNumber = changeNumber;
+        mUpdateTimestamp = updateTimestamp;
+    }
+
+    public Set<RuleBasedSegment> getActive() {
+        return mActive;
+    }
+
+    public Set<RuleBasedSegment> getArchived() {
+        return mArchived;
+    }
+
+    public long getChangeNumber() {
+        return mChangeNumber;
+    }
+
+    public long getUpdateTimestamp() {
+        return mUpdateTimestamp;
+    }
+}

src/main/java/io/split/android/client/service/rules/RuleBasedSegmentChangeProcessor.java

@@ -0,0 +1,30 @@
+package io.split.android.client.service.rules;
+
+import java.util.Collections;
+import java.util.HashSet;
+import java.util.List;
+import java.util.Set;
+
+import io.split.android.client.dtos.RuleBasedSegment;
+import io.split.android.client.dtos.Status;
+
+public class RuleBasedSegmentChangeProcessor {
+
+    public ProcessedRuleBasedSegmentChange process(List<RuleBasedSegment> segments, long changeNumber) {
+        Set<RuleBasedSegment> toAdd = new HashSet<>();
+        Set<RuleBasedSegment> toRemove = new HashSet<>();
+        for (RuleBasedSegment segment : segments) {
+            if (segment.getStatus() == Status.ACTIVE) {
+                toAdd.add(segment);
+            } else {
+                toRemove.add(segment);
+            }
+        }
+
+        return new ProcessedRuleBasedSegmentChange(toAdd, toRemove, changeNumber, System.currentTimeMillis());
+    }
+
+    public ProcessedRuleBasedSegmentChange process(RuleBasedSegment segment, long changeNumber) {
+        return process(Collections.singletonList(segment), changeNumber);
+    }
+}

src/main/java/io/split/android/client/service/splits/RuleBasedSegmentInPlaceUpdateTask.java

@@ -0,0 +1,57 @@
+package io.split.android.client.service.splits;
+
+import static io.split.android.client.utils.Utils.checkNotNull;
+
+import androidx.annotation.NonNull;
+
+import io.split.android.client.dtos.RuleBasedSegment;
+import io.split.android.client.events.ISplitEventsManager;
+import io.split.android.client.events.SplitInternalEvent;
+import io.split.android.client.service.executor.SplitTask;
+import io.split.android.client.service.executor.SplitTaskExecutionInfo;
+import io.split.android.client.service.executor.SplitTaskType;
+import io.split.android.client.service.rules.ProcessedRuleBasedSegmentChange;
+import io.split.android.client.service.rules.RuleBasedSegmentChangeProcessor;
+import io.split.android.client.storage.rbs.RuleBasedSegmentStorage;
+import io.split.android.client.utils.logger.Logger;
+
+public class RuleBasedSegmentInPlaceUpdateTask implements SplitTask {
+
+    private final RuleBasedSegmentStorage mRuleBasedSegmentStorage;
+    private final long mChangeNumber;
+    private final RuleBasedSegment mRuleBasedSegment;
+    private final RuleBasedSegmentChangeProcessor mChangeProcessor;
+    private final ISplitEventsManager mEventsManager;
+
+    public RuleBasedSegmentInPlaceUpdateTask(@NonNull RuleBasedSegmentStorage ruleBasedSegmentStorage,
+                                             @NonNull RuleBasedSegmentChangeProcessor changeProcessor,
+                                             @NonNull ISplitEventsManager eventsManager,
+                                             @NonNull RuleBasedSegment ruleBasedSegment,
+                                             long changeNumber) {
+        mRuleBasedSegmentStorage = checkNotNull(ruleBasedSegmentStorage);
+        mRuleBasedSegment = checkNotNull(ruleBasedSegment);
+        mChangeProcessor = checkNotNull(changeProcessor);
+        mEventsManager = eventsManager;
+        mChangeNumber = changeNumber;
+    }
+
+    @NonNull
+    @Override
+    public SplitTaskExecutionInfo execute() {
+        try {
+            ProcessedRuleBasedSegmentChange processedChange = mChangeProcessor.process(mRuleBasedSegment, mChangeNumber);
+            boolean triggerSdkUpdate = mRuleBasedSegmentStorage.update(processedChange.getActive(), processedChange.getArchived(), mChangeNumber);
+
+            if (triggerSdkUpdate) {
+                mEventsManager.notifyInternalEvent(SplitInternalEvent.RULE_BASED_SEGMENTS_UPDATED);
+            }
+
+            Logger.v("Updated rule based segment");
+            return SplitTaskExecutionInfo.success(SplitTaskType.RULE_BASED_SEGMENT_SYNC);
+        } catch (Exception ex) {
+            Logger.e("Could not update rule based segment");
+
+            return SplitTaskExecutionInfo.error(SplitTaskType.RULE_BASED_SEGMENT_SYNC);
+        }
+    }
+}

src/main/java/io/split/android/client/service/splits/SplitsSyncHelper.java

@@ -8,16 +8,12 @@
 import androidx.annotation.VisibleForTesting;
 
 import java.util.Collections;
-import java.util.HashSet;
 import java.util.LinkedHashMap;
 import java.util.Map;
-import java.util.Set;
 import java.util.concurrent.TimeUnit;
 
-import io.split.android.client.dtos.RuleBasedSegment;
 import io.split.android.client.dtos.RuleBasedSegmentChange;
 import io.split.android.client.dtos.SplitChange;
-import io.split.android.client.dtos.Status;
 import io.split.android.client.dtos.TargetingRulesChange;
 import io.split.android.client.network.SplitHttpHeadersBuilder;
 import io.split.android.client.service.ServiceConstants;
@@ -26,6 +22,8 @@
 import io.split.android.client.service.http.HttpFetcher;
 import io.split.android.client.service.http.HttpFetcherException;
 import io.split.android.client.service.http.HttpStatus;
+import io.split.android.client.service.rules.ProcessedRuleBasedSegmentChange;
+import io.split.android.client.service.rules.RuleBasedSegmentChangeProcessor;
 import io.split.android.client.service.sseclient.BackoffCounter;
 import io.split.android.client.service.sseclient.ReconnectBackoffCounter;
 import io.split.android.client.storage.rbs.RuleBasedSegmentStorageProducer;
@@ -44,6 +42,7 @@ public class SplitsSyncHelper {
     private final HttpFetcher<TargetingRulesChange> mSplitFetcher;
     private final SplitsStorage mSplitsStorage;
     private final SplitChangeProcessor mSplitChangeProcessor;
+    private final RuleBasedSegmentChangeProcessor mRuleBasedSegmentChangeProcessor;
     private final RuleBasedSegmentStorageProducer mRuleBasedSegmentStorage;
     private final TelemetryRuntimeProducer mTelemetryRuntimeProducer;
     private final BackoffCounter mBackoffCounter;
@@ -52,12 +51,14 @@ public class SplitsSyncHelper {
     public SplitsSyncHelper(@NonNull HttpFetcher<TargetingRulesChange> splitFetcher,
                             @NonNull SplitsStorage splitsStorage,
                             @NonNull SplitChangeProcessor splitChangeProcessor,
+                            @NonNull RuleBasedSegmentChangeProcessor ruleBasedSegmentChangeProcessor,
                             @NonNull RuleBasedSegmentStorageProducer ruleBasedSegmentStorage,
                             @NonNull TelemetryRuntimeProducer telemetryRuntimeProducer,
                             @Nullable String flagsSpec) {
         this(splitFetcher,
                 splitsStorage,
                 splitChangeProcessor,
+                ruleBasedSegmentChangeProcessor,
                 ruleBasedSegmentStorage,
                 telemetryRuntimeProducer,
                 new ReconnectBackoffCounter(1, ON_DEMAND_FETCH_BACKOFF_MAX_WAIT),
@@ -68,13 +69,15 @@ public SplitsSyncHelper(@NonNull HttpFetcher<TargetingRulesChange> splitFetcher,
     public SplitsSyncHelper(@NonNull HttpFetcher<TargetingRulesChange> splitFetcher,
                             @NonNull SplitsStorage splitsStorage,
                             @NonNull SplitChangeProcessor splitChangeProcessor,
+                            @NonNull RuleBasedSegmentChangeProcessor ruleBasedSegmentChangeProcessor,
                             @NonNull RuleBasedSegmentStorageProducer ruleBasedSegmentStorage,
                             @NonNull TelemetryRuntimeProducer telemetryRuntimeProducer,
                             @NonNull BackoffCounter backoffCounter,
                             @Nullable String flagsSpec) {
         mSplitFetcher = checkNotNull(splitFetcher);
         mSplitsStorage = checkNotNull(splitsStorage);
         mSplitChangeProcessor = checkNotNull(splitChangeProcessor);
+        mRuleBasedSegmentChangeProcessor = checkNotNull(ruleBasedSegmentChangeProcessor);
         mRuleBasedSegmentStorage = checkNotNull(ruleBasedSegmentStorage);
         mTelemetryRuntimeProducer = checkNotNull(telemetryRuntimeProducer);
         mBackoffCounter = checkNotNull(backoffCounter);
@@ -205,17 +208,8 @@ private void updateStorage(boolean clearBeforeUpdate, SplitChange splitChange, R
     }
 
     private void updateRbsStorage(RuleBasedSegmentChange ruleBasedSegmentChange) {
-        long changeNumber = ruleBasedSegmentChange.getTill();
-        Set<RuleBasedSegment> toAdd = new HashSet<>();
-        Set<RuleBasedSegment> toRemove = new HashSet<>();
-        for (RuleBasedSegment segment : ruleBasedSegmentChange.getSegments()) {
-            if (segment.getStatus() == Status.ACTIVE) {
-                toAdd.add(segment);
-            } else {
-                toRemove.add(segment);
-            }
-        }
-        mRuleBasedSegmentStorage.update(toAdd, toRemove, changeNumber);
+        ProcessedRuleBasedSegmentChange change = mRuleBasedSegmentChangeProcessor.process(ruleBasedSegmentChange.getSegments(), ruleBasedSegmentChange.getTill());
+        mRuleBasedSegmentStorage.update(change.getActive(), change.getArchived(), change.getChangeNumber());
     }
 
     private void logError(String message) {