Merge branch 'feature/rule-based-segment' into rbs-localhost

Author: chillaq

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

@@ -220,7 +220,6 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         // Segments
         _segmentSynchronizationTaskImp = buildSegments(config, segmentCache, splitCache);
 
-
         SplitParser splitParser = new SplitParser();
         RuleBasedSegmentParser ruleBasedSegmentParser = new RuleBasedSegmentParser();
         // SplitFetcher

client/src/main/java/io/split/engine/common/SynchronizerImp.java

@@ -118,8 +118,8 @@ private SyncResult attemptSplitsSync(long targetChangeNumber, long ruleBasedSegm
             if (fetchResult != null && !fetchResult.retry() && !fetchResult.isSuccess()) {
                 return new SyncResult(false, remainingAttempts, fetchResult);
             }
-            if ((targetChangeNumber != 0 && targetChangeNumber <= _splitCacheProducer.getChangeNumber()) ||
-                    (ruleBasedSegmentChangeNumber != 0 && ruleBasedSegmentChangeNumber <= _ruleBasedSegmentCacheProducer.getChangeNumber())) {
+            if (targetChangeNumber <= _splitCacheProducer.getChangeNumber()
+                    && ruleBasedSegmentChangeNumber <= _ruleBasedSegmentCacheProducer.getChangeNumber()) {
                 return new SyncResult(true, remainingAttempts, fetchResult);
             } else if (remainingAttempts <= 0) {
                 return new SyncResult(false, remainingAttempts, fetchResult);
@@ -137,9 +137,15 @@ private SyncResult attemptSplitsSync(long targetChangeNumber, long ruleBasedSegm
     @Override
     public void refreshSplits(Long targetChangeNumber, Long ruleBasedSegmentChangeNumber) {
 
-        if ((targetChangeNumber != 0 && targetChangeNumber <= _splitCacheProducer.getChangeNumber()) ||
-                (ruleBasedSegmentChangeNumber != 0 && ruleBasedSegmentChangeNumber <= _ruleBasedSegmentCacheProducer.getChangeNumber()) ||
-                (ruleBasedSegmentChangeNumber == 0 && targetChangeNumber == 0)) {
+        if (targetChangeNumber == null || targetChangeNumber == 0) {
+            targetChangeNumber = _splitCacheProducer.getChangeNumber();
+        }
+        if (ruleBasedSegmentChangeNumber == null || ruleBasedSegmentChangeNumber == 0) {
+            ruleBasedSegmentChangeNumber = _ruleBasedSegmentCacheProducer.getChangeNumber();
+        }
+        
+        if (targetChangeNumber <= _splitCacheProducer.getChangeNumber()
+                && ruleBasedSegmentChangeNumber <= _ruleBasedSegmentCacheProducer.getChangeNumber()) {
             return;
         }
 

client/src/main/java/io/split/engine/experiments/SplitFetcherImp.java

@@ -144,7 +144,7 @@ private Set<String> runWithoutExceptionHandling(FetchOptions options) throws Int
 
         synchronized (_lock) {
             // check state one more time.
-            if (checkExitConditions(change.featureFlags, _splitCacheProducer.getChangeNumber()) &&
+            if (checkExitConditions(change.featureFlags, _splitCacheProducer.getChangeNumber()) ||
                     checkExitConditions(change.ruleBasedSegments, _ruleBasedSegmentCacheProducer.getChangeNumber())) {
                 // some other thread may have updated the shared state. exit
                 return segments;

client/src/main/java/io/split/engine/sse/NotificationParserImp.java

@@ -1,17 +1,18 @@
 package io.split.engine.sse;
 
+import io.split.client.dtos.RuleBasedSegment;
+import io.split.client.dtos.Split;
 import io.split.client.utils.Json;
 
 import io.split.engine.sse.dtos.ControlNotification;
 import io.split.engine.sse.dtos.ErrorNotification;
-import io.split.engine.sse.dtos.FeatureFlagChangeNotification;
+import io.split.engine.sse.dtos.CommonChangeNotification;
 import io.split.engine.sse.dtos.GenericNotificationData;
 import io.split.engine.sse.dtos.IncomingNotification;
 import io.split.engine.sse.dtos.OccupancyNotification;
 import io.split.engine.sse.dtos.RawMessageNotification;
 import io.split.engine.sse.dtos.SegmentChangeNotification;
 import io.split.engine.sse.dtos.SplitKillNotification;
-import io.split.engine.sse.dtos.RuleBasedSegmentChangeNotification;
 import io.split.engine.sse.exceptions.EventParsingException;
 
 public class NotificationParserImp implements NotificationParser {
@@ -48,9 +49,9 @@ public ErrorNotification parseError(String payload) throws EventParsingException
     private IncomingNotification parseNotification(GenericNotificationData genericNotificationData) throws Exception {
         switch (genericNotificationData.getType()) {
             case SPLIT_UPDATE:
-                return new FeatureFlagChangeNotification(genericNotificationData);
+                return new CommonChangeNotification(genericNotificationData, Split.class);
             case RB_SEGMENT_UPDATE:
-                return new RuleBasedSegmentChangeNotification(genericNotificationData);
+                return new CommonChangeNotification(genericNotificationData, RuleBasedSegment.class);
             case SPLIT_KILL:
                 return new SplitKillNotification(genericNotificationData);
             case SEGMENT_UPDATE:

client/src/main/java/io/split/engine/sse/NotificationProcessor.java

@@ -1,15 +1,12 @@
 package io.split.engine.sse;
 
-import io.split.engine.sse.dtos.FeatureFlagChangeNotification;
 import io.split.engine.sse.dtos.IncomingNotification;
 import io.split.engine.sse.dtos.SplitKillNotification;
 import io.split.engine.sse.dtos.StatusNotification;
-import io.split.engine.sse.dtos.RuleBasedSegmentChangeNotification;
 
 public interface NotificationProcessor {
     void process(IncomingNotification notification);
-    void processSplitUpdate(FeatureFlagChangeNotification featureFlagChangeNotification);
-    void processRuleBasedSegmentUpdate(RuleBasedSegmentChangeNotification ruleBasedSegmentChangeNotification);
+    void processUpdates(IncomingNotification notification);
     void processSplitKill(SplitKillNotification splitKillNotification);
     void processSegmentUpdate(long changeNumber, String segmentName);
     void processStatus(StatusNotification statusNotification);

client/src/main/java/io/split/engine/sse/NotificationProcessorImp.java

@@ -1,13 +1,11 @@
 package io.split.engine.sse;
 
 import com.google.common.annotations.VisibleForTesting;
-import io.split.engine.sse.dtos.FeatureFlagChangeNotification;
 import io.split.engine.sse.dtos.GenericNotificationData;
 import io.split.engine.sse.dtos.IncomingNotification;
 import io.split.engine.sse.dtos.SplitKillNotification;
 import io.split.engine.sse.dtos.StatusNotification;
 import io.split.engine.sse.dtos.SegmentQueueDto;
-import io.split.engine.sse.dtos.RuleBasedSegmentChangeNotification;
 import io.split.engine.sse.workers.FeatureFlagsWorker;
 import io.split.engine.sse.workers.Worker;
 
@@ -32,25 +30,19 @@ public static NotificationProcessorImp build(FeatureFlagsWorker featureFlagsWork
         return new NotificationProcessorImp(featureFlagsWorker, segmentWorker, pushStatusTracker);
     }
 
-    @Override
-    public void process(IncomingNotification notification) {
-        notification.handler(this);
-    }
-
-    @Override
-    public void processSplitUpdate(FeatureFlagChangeNotification featureFlagChangeNotification) {
-        _featureFlagsWorker.addToQueue(featureFlagChangeNotification);
+    public void processUpdates(IncomingNotification notification) {
+        _featureFlagsWorker.addToQueue(notification);
     }
 
     @Override
-    public void processRuleBasedSegmentUpdate(RuleBasedSegmentChangeNotification ruleBasedSegmentChangeNotification) {
-        _featureFlagsWorker.addToQueue(ruleBasedSegmentChangeNotification);
+    public void process(IncomingNotification notification) {
+        notification.handler(this);
     }
 
     @Override
     public void processSplitKill(SplitKillNotification splitKillNotification) {
         _featureFlagsWorker.kill(splitKillNotification);
-        _featureFlagsWorker.addToQueue(new FeatureFlagChangeNotification(GenericNotificationData.builder()
+        _featureFlagsWorker.addToQueue(new SplitKillNotification(GenericNotificationData.builder()
                 .changeNumber(splitKillNotification.getChangeNumber())
                 .channel(splitKillNotification.getChannel())
                 .build()));

client/src/main/java/io/split/engine/sse/dtos/CommonChangeNotification.java

@@ -1,5 +1,6 @@
 package io.split.engine.sse.dtos;
 
+import io.split.client.utils.Json;
 import io.split.engine.segments.SegmentSynchronizationTaskImp;
 import io.split.engine.sse.NotificationProcessor;
 import io.split.engine.sse.enums.CompressType;
@@ -14,24 +15,29 @@
 import static io.split.engine.sse.utils.DecompressionUtil.gZipDecompress;
 import static io.split.engine.sse.utils.DecompressionUtil.zLibDecompress;
 
-public class CommonChangeNotification extends IncomingNotification {
+public class CommonChangeNotification<Y> extends IncomingNotification {
     private static final Logger _log = LoggerFactory.getLogger(SegmentSynchronizationTaskImp.class);
     private final long changeNumber;
     private long previousChangeNumber;
     private CompressType compressType;
+    private Y definition;
+    private Class _definitionClass;
 
-    public CommonChangeNotification(GenericNotificationData genericNotificationData,  IncomingNotification.Type notificationType) {
-        super(notificationType, genericNotificationData.getChannel());
+    public CommonChangeNotification(GenericNotificationData genericNotificationData,
+                                    Class definitionClass) {
+        super(genericNotificationData.getType(), genericNotificationData.getChannel());
         changeNumber = genericNotificationData.getChangeNumber();
+        _definitionClass = definitionClass;
+
         if(genericNotificationData.getPreviousChangeNumber() != null) {
             previousChangeNumber = genericNotificationData.getPreviousChangeNumber();
         }
         compressType =  CompressType.from(genericNotificationData.getCompressType());
-        if (compressType == null || genericNotificationData.getFeatureFlagDefinition() == null) {
+        if (compressType == null || genericNotificationData.getDefinition() == null) {
             return;
         }
         try {
-            byte[] decodedBytes = Base64.getDecoder().decode(genericNotificationData.getFeatureFlagDefinition());
+            byte[] decodedBytes = Base64.getDecoder().decode(genericNotificationData.getDefinition());
             switch (compressType) {
                 case GZIP:
                     decodedBytes = gZipDecompress(decodedBytes);
@@ -57,18 +63,25 @@ public long getChangeNumber() {
     public long getPreviousChangeNumber() {
         return previousChangeNumber;
     }
-
     public CompressType getCompressType() {
         return compressType;
     }
 
+    public Y getDefinition() {
+        return definition;
+    }
+
     @Override
-    public void handler(NotificationProcessor notificationProcessor) {}
+    public void handler(NotificationProcessor notificationProcessor) {
+        notificationProcessor.processUpdates(this);
+    }
 
     @Override
     public String toString() {
         return String.format("Type: %s; Channel: %s; ChangeNumber: %s", getType(), getChannel(), getChangeNumber());
     }
 
-    public void updateDefinition(byte[] decodedBytes) throws UnsupportedEncodingException {};
+    private void updateDefinition(byte[] decodedBytes) throws UnsupportedEncodingException {
+        definition = (Y) Json.fromJson(new String(decodedBytes, "UTF-8"), _definitionClass);
+    }
 }

client/src/main/java/io/split/engine/sse/dtos/FeatureFlagChangeNotification.java

@@ -75,7 +75,7 @@ public Long getPreviousChangeNumber() {
         return previousChangeNumber;
     }
 
-    public String getFeatureFlagDefinition() {
+    public String getDefinition() {
         return featureFlagDefinition;
     }