Added dto and updated client and engine classes

Author: chillaq

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

@@ -6,6 +6,8 @@ private Spec() {
         // restrict instantiation
     }
 
-    public static final String SPEC_VERSION = "1.1";
+    public static String SPEC_VERSION = "1.3";
+    public static final String SPEC_1_3 = "1.3";
+    public static final String SPEC_1_1 = "1.1";
 }
 

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

@@ -2,6 +2,7 @@
 
 import com.google.common.annotations.VisibleForTesting;
 
+import io.split.Spec;
 import io.split.client.dtos.SplitChange;
 import io.split.client.dtos.SplitHttpResponse;
 import io.split.client.exceptions.UriTooLongException;
@@ -31,6 +32,7 @@ public final class HttpSplitChangeFetcher implements SplitChangeFetcher {
     private static final Logger _log = LoggerFactory.getLogger(HttpSplitChangeFetcher.class);
 
     private static final String SINCE = "since";
+    private static final String RB_SINCE = "rbSince";
     private static final String TILL = "till";
     private static final String SETS = "sets";
     private static final String SPEC = "s";
@@ -56,38 +58,48 @@ long makeRandomTill() {
     }
 
     @Override
-    public SplitChange fetch(long since, FetchOptions options) {
+    public SplitChange fetch(long since, long sinceRBS, FetchOptions options) {
 
         long start = System.currentTimeMillis();
+        for (int i=0; i<2; i++) {
+            try {
+                URIBuilder uriBuilder = new URIBuilder(_target).addParameter(SPEC, "" + SPEC_VERSION);
+                uriBuilder.addParameter(SINCE, "" + since);
+                if (SPEC_VERSION.equals(Spec.SPEC_1_3)) {
+                    uriBuilder.addParameter(RB_SINCE, "" + sinceRBS);
+                }
+                if (!options.flagSetsFilter().isEmpty()) {
+                    uriBuilder.addParameter(SETS, "" + options.flagSetsFilter());
+                }
+                if (options.hasCustomCN()) {
+                    uriBuilder.addParameter(TILL, "" + options.targetCN());
+                }
+                URI uri = uriBuilder.build();
+                SplitHttpResponse response = _client.get(uri, options, null);
 
-        try {
-            URIBuilder uriBuilder = new URIBuilder(_target).addParameter(SPEC, "" + SPEC_VERSION);
-            uriBuilder.addParameter(SINCE, "" + since);
-            if (!options.flagSetsFilter().isEmpty()) {
-                uriBuilder.addParameter(SETS, "" + options.flagSetsFilter());
-            }
-            if (options.hasCustomCN()) {
-                uriBuilder.addParameter(TILL, "" + options.targetCN());
-            }
-            URI uri = uriBuilder.build();
-            SplitHttpResponse response = _client.get(uri, options, null);
-
-            if (response.statusCode() < HttpStatus.SC_OK || response.statusCode() >= HttpStatus.SC_MULTIPLE_CHOICES) {
-                if (response.statusCode() == HttpStatus.SC_REQUEST_URI_TOO_LONG) {
-                    _log.error("The amount of flag sets provided are big causing uri length error.");
-                    throw new UriTooLongException(String.format("Status code: %s. Message: %s", response.statusCode(), response.statusMessage()));
+                if (response.statusCode() < HttpStatus.SC_OK || response.statusCode() >= HttpStatus.SC_MULTIPLE_CHOICES) {
+                    if (response.statusCode() == HttpStatus.SC_REQUEST_URI_TOO_LONG) {
+                        _log.error("The amount of flag sets provided are big causing uri length error.");
+                        throw new UriTooLongException(String.format("Status code: %s. Message: %s", response.statusCode(), response.statusMessage()));
+                    }
+                    if (response.statusCode() == HttpStatus.SC_BAD_REQUEST && response.statusMessage().equals("unknown spec")) {
+                        _log.warn(String.format("Detected old spec response, falling back to spec 1.1"));
+                        SPEC_VERSION = Spec.SPEC_1_1;
+                        continue;
+                    }
+                    _telemetryRuntimeProducer.recordSyncError(ResourceEnum.SPLIT_SYNC, response.statusCode());
+                    throw new IllegalStateException(
+                            String.format("Could not retrieve splitChanges since %s; http return code %s", since, response.statusCode())
+                    );
                 }
-                _telemetryRuntimeProducer.recordSyncError(ResourceEnum.SPLIT_SYNC, response.statusCode());
-                throw new IllegalStateException(
-                        String.format("Could not retrieve splitChanges since %s; http return code %s", since, response.statusCode())
-                );
+                return Json.fromJson(response.body(), SplitChange.class);
+            } catch (Exception e) {
+                throw new IllegalStateException(String.format("Problem fetching splitChanges since %s: %s", since, e), e);
+            } finally {
+                _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SPLITS, System.currentTimeMillis() - start);
             }
-            return Json.fromJson(response.body(), SplitChange.class);
-        } catch (Exception e) {
-            throw new IllegalStateException(String.format("Problem fetching splitChanges since %s: %s", since, e), e);
-        } finally {
-            _telemetryRuntimeProducer.recordSyncLatency(HTTPLatenciesEnum.SPLITS, System.currentTimeMillis()-start);
         }
+        return null;
     }
 
     @VisibleForTesting

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

@@ -29,7 +29,7 @@ public JsonLocalhostSplitChangeFetcher(InputStreamProvider inputStreamProvider)
     }
 
     @Override
-    public SplitChange fetch(long since, FetchOptions options) {
+    public SplitChange fetch(long since, long RBsince,  FetchOptions options) {
         try {
             JsonReader jsonReader = new JsonReader(new BufferedReader(new InputStreamReader(_inputStreamProvider.get(), StandardCharsets.UTF_8)));
             SplitChange splitChange = Json.fromJson(jsonReader, SplitChange.class);

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

@@ -34,7 +34,7 @@ public LegacyLocalhostSplitChangeFetcher(String directory) {
     }
 
     @Override
-    public SplitChange fetch(long since, FetchOptions options) {
+    public SplitChange fetch(long since, long sinceRBS, FetchOptions options) {
 
         try (BufferedReader reader = new BufferedReader(new FileReader(_splitFile))) {
             SplitChange splitChange = new SplitChange();

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

@@ -53,6 +53,7 @@
 import io.split.engine.experiments.SplitFetcher;
 import io.split.engine.experiments.SplitFetcherImp;
 import io.split.engine.experiments.SplitParser;
+import io.split.engine.experiments.RuleBasedSegmentParser;
 import io.split.engine.experiments.SplitSynchronizationTask;
 import io.split.engine.segments.SegmentChangeFetcher;
 import io.split.engine.segments.SegmentSynchronizationTaskImp;
@@ -66,8 +67,11 @@
 import io.split.storages.SplitCache;
 import io.split.storages.SplitCacheConsumer;
 import io.split.storages.SplitCacheProducer;
+import io.split.storages.RuleBasedSegmentCacheConsumer;
+import io.split.storages.RuleBasedSegmentCache;
 import io.split.storages.enums.OperationMode;
 import io.split.storages.memory.InMemoryCacheImp;
+import io.split.storages.memory.RuleBasedSegmentCacheInMemoryImp;
 import io.split.storages.memory.SegmentCacheInMemoryImpl;
 import io.split.storages.pluggable.adapters.UserCustomEventAdapterProducer;
 import io.split.storages.pluggable.adapters.UserCustomImpressionAdapterConsumer;
@@ -123,8 +127,7 @@
 public class SplitFactoryImpl implements SplitFactory {
     private static final org.slf4j.Logger _log = LoggerFactory.getLogger(SplitFactoryImpl.class);
     private static final String LEGACY_LOG_MESSAGE = "The sdk initialize in localhost mode using Legacy file. The splitFile or "
-            +
-            "inputStream doesn't add it to the config.";
+            + "inputStream were not added to the config.";
     private final static long SSE_CONNECT_TIMEOUT = 30000;
     private final static long SSE_SOCKET_TIMEOUT = 70000;
 
@@ -204,6 +207,7 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         SegmentCache segmentCache = new SegmentCacheInMemoryImpl();
         FlagSetsFilter flagSetsFilter = new FlagSetsFilterImpl(config.getSetsFilter());
         SplitCache splitCache = new InMemoryCacheImp(flagSetsFilter);
+        RuleBasedSegmentCache ruleBasedSegmentCache = new RuleBasedSegmentCacheInMemoryImp();
         ImpressionsStorage impressionsStorage = new InMemoryImpressionsStorage(config.impressionsQueueSize());
         _splitCache = splitCache;
         _segmentCache = segmentCache;
@@ -215,8 +219,11 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         _segmentSynchronizationTaskImp = buildSegments(config, segmentCache, splitCache);
 
         SplitParser splitParser = new SplitParser();
+        RuleBasedSegmentParser ruleBasedSegmentParser = new RuleBasedSegmentParser();
+
         // SplitFetcher
-        _splitFetcher = buildSplitFetcher(splitCache, splitParser, flagSetsFilter);
+        _splitFetcher = buildSplitFetcher(splitCache, splitParser, flagSetsFilter, ruleBasedSegmentParser, ruleBasedSegmentCache);
+
 
         // SplitSynchronizationTask
         _splitSynchronizationTask = new SplitSynchronizationTask(_splitFetcher,
@@ -244,7 +251,7 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
                 config.getThreadFactory());
 
         // Evaluator
-        _evaluator = new EvaluatorImp(splitCache, segmentCache);
+        _evaluator = new EvaluatorImp(splitCache, segmentCache, ruleBasedSegmentCache);
 
         // SplitClient
         _client = new SplitClientImpl(this,
@@ -301,6 +308,9 @@ protected SplitFactoryImpl(String apiToken, SplitClientConfig config, CustomStor
                 customStorageWrapper);
         UserCustomSplitAdapterConsumer userCustomSplitAdapterConsumer = new UserCustomSplitAdapterConsumer(
                 customStorageWrapper);
+        // TODO Update the instance to UserCustomRuleBasedSegmentAdapterConsumer
+        RuleBasedSegmentCacheConsumer userCustomRuleBasedSegmentAdapterConsumer = new RuleBasedSegmentCacheInMemoryImp();
+
         // TODO migrate impressions sender to Task instead manager and not instantiate
         // Producer here.
         UserCustomImpressionAdapterConsumer userCustomImpressionAdapterConsumer = new UserCustomImpressionAdapterConsumer();
@@ -333,7 +343,7 @@ protected SplitFactoryImpl(String apiToken, SplitClientConfig config, CustomStor
         _gates = new SDKReadinessGates();
 
         _telemetrySynchronizer = new TelemetryConsumerSubmitter(customStorageWrapper, _sdkMetadata);
-        _evaluator = new EvaluatorImp(userCustomSplitAdapterConsumer, userCustomSegmentAdapterConsumer);
+        _evaluator = new EvaluatorImp(userCustomSplitAdapterConsumer, userCustomSegmentAdapterConsumer, userCustomRuleBasedSegmentAdapterConsumer);
         _impressionsSender = PluggableImpressionSender.create(customStorageWrapper);
         _uniqueKeysTracker = createUniqueKeysTracker(config);
         _impressionsManager = buildImpressionsManager(config, userCustomImpressionAdapterConsumer,
@@ -391,6 +401,7 @@ protected SplitFactoryImpl(SplitClientConfig config) {
 
         SegmentCache segmentCache = new SegmentCacheInMemoryImpl();
         FlagSetsFilter flagSetsFilter = new FlagSetsFilterImpl(config.getSetsFilter());
+        RuleBasedSegmentCache ruleBasedSegmentCache = new RuleBasedSegmentCacheInMemoryImp();
         SplitCache splitCache = new InMemoryCacheImp(flagSetsFilter);
         _splitCache = splitCache;
         _gates = new SDKReadinessGates();
@@ -413,10 +424,11 @@ protected SplitFactoryImpl(SplitClientConfig config) {
 
         // SplitFetcher
         SplitChangeFetcher splitChangeFetcher = createSplitChangeFetcher(config);
+        RuleBasedSegmentParser ruleBasedSegmentParser = new RuleBasedSegmentParser();
         SplitParser splitParser = new SplitParser();
 
         _splitFetcher = new SplitFetcherImp(splitChangeFetcher, splitParser, splitCache, _telemetryStorageProducer,
-                flagSetsFilter);
+                flagSetsFilter, ruleBasedSegmentParser, ruleBasedSegmentCache);
 
         // SplitSynchronizationTask
         _splitSynchronizationTask = new SplitSynchronizationTask(_splitFetcher, splitCache,
@@ -428,7 +440,7 @@ protected SplitFactoryImpl(SplitClientConfig config) {
                 _impressionsManager, null, null, null);
 
         // Evaluator
-        _evaluator = new EvaluatorImp(splitCache, segmentCache);
+        _evaluator = new EvaluatorImp(splitCache, segmentCache, ruleBasedSegmentCache);
 
         EventsStorage eventsStorage = new NoopEventsStorageImp();
 
@@ -609,11 +621,12 @@ private SegmentSynchronizationTaskImp buildSegments(SplitClientConfig config,
     }
 
     private SplitFetcher buildSplitFetcher(SplitCacheProducer splitCacheProducer, SplitParser splitParser,
-            FlagSetsFilter flagSetsFilter) throws URISyntaxException {
+                                           FlagSetsFilter flagSetsFilter, RuleBasedSegmentParser ruleBasedSegmentParser,
+                                           RuleBasedSegmentCache ruleBasedSegmentCache) throws URISyntaxException {
         SplitChangeFetcher splitChangeFetcher = HttpSplitChangeFetcher.create(_splitHttpClient, _rootTarget,
                 _telemetryStorageProducer);
         return new SplitFetcherImp(splitChangeFetcher, splitParser, splitCacheProducer, _telemetryStorageProducer,
-                flagSetsFilter);
+                flagSetsFilter,ruleBasedSegmentParser, ruleBasedSegmentCache);
     }
 
     private ImpressionsManagerImpl buildImpressionsManager(SplitClientConfig config,

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

@@ -32,7 +32,7 @@ public YamlLocalhostSplitChangeFetcher(InputStreamProvider inputStreamProvider)
     }
 
     @Override
-    public SplitChange fetch(long since, FetchOptions options) {
+    public SplitChange fetch(long since, long sinceRBS, FetchOptions options) {
         try {
             Yaml yaml = new Yaml();
             List<Map<String, Map<String, Object>>> yamlSplits = yaml.load(_inputStreamProvider.get());

client/src/main/java/io/split/client/dtos/Excluded.java

@@ -0,0 +1,8 @@
+package io.split.client.dtos;
+
+import java.util.List;
+
+public class Excluded {
+    public List<String> keys;
+    public List<String> segments;
+}

client/src/main/java/io/split/client/dtos/RuleBasedSegment.java

@@ -0,0 +1,12 @@
+package io.split.client.dtos;
+
+import java.util.List;
+
+public class RuleBasedSegment {
+    public String name;
+    public Status status;
+    public String trafficTypeName;
+    public long changeNumber;
+    public List<Condition> conditions;
+    public Excluded excluded;
+}

client/src/main/java/io/split/client/dtos/RuleBasedSegmentChange.java

@@ -0,0 +1,9 @@
+package io.split.client.dtos;
+
+import java.util.List;
+
+public class RuleBasedSegmentChange {
+    public List<RuleBasedSegment> ruleBasedSegments;
+    public long since;
+    public long till;
+}

client/src/main/java/io/split/client/dtos/SplitChange.java

@@ -6,4 +6,7 @@ public class SplitChange {
     public List<Split> splits;
     public long since;
     public long till;
+    public List<RuleBasedSegment> ruleBasedSegments;
+    public long sinceRBS;
+    public long tillRBS;
 }