Updated Splitfetcher classes

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

@@ -2,9 +2,15 @@
 
 import com.google.common.annotations.VisibleForTesting;
 
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import io.split.Spec;
+import io.split.client.dtos.RuleBasedSegment;
+import io.split.client.dtos.Split;
 import io.split.client.dtos.SplitChange;
 import io.split.client.dtos.SplitHttpResponse;
 import io.split.client.exceptions.UriTooLongException;
+import io.split.client.utils.GenericClientUtil;
 import io.split.client.utils.Json;
 import io.split.client.utils.Utils;
 import io.split.engine.common.FetchOptions;
@@ -20,6 +26,7 @@
 
 import java.net.URI;
 import java.net.URISyntaxException;
+import java.util.ArrayList;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static io.split.Spec.SPEC_VERSION;
@@ -31,6 +38,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 +64,50 @@ 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())
+                    );
+                }
+                if (SPEC_VERSION.equals(Spec.SPEC_1_1)) {
+                    return Json.fromJson(response.body(), SplitChange.class);
                 }
-                _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 GenericClientUtil.ExtractFeatureFlagsAndRuleBasedSegments(response.body());
+            } 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 sinceRBS, 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

@@ -54,6 +54,7 @@
 import io.split.engine.experiments.SplitFetcherImp;
 import io.split.engine.experiments.SplitParser;
 import io.split.engine.experiments.SplitSynchronizationTask;
+import io.split.engine.experiments.RuleBasedSegmentParser;
 import io.split.engine.segments.SegmentChangeFetcher;
 import io.split.engine.segments.SegmentSynchronizationTaskImp;
 import io.split.integrations.IntegrationsConfig;
@@ -68,6 +69,7 @@
 import io.split.storages.SplitCacheProducer;
 import io.split.storages.RuleBasedSegmentCacheConsumer;
 import io.split.storages.RuleBasedSegmentCache;
+import io.split.storages.RuleBasedSegmentCacheProducer;
 import io.split.storages.enums.OperationMode;
 import io.split.storages.memory.InMemoryCacheImp;
 import io.split.storages.memory.SegmentCacheInMemoryImpl;
@@ -220,8 +222,10 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
 
 
         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,
@@ -422,9 +426,10 @@ protected SplitFactoryImpl(SplitClientConfig config) {
         // SplitFetcher
         SplitChangeFetcher splitChangeFetcher = createSplitChangeFetcher(config);
         SplitParser splitParser = new SplitParser();
+        RuleBasedSegmentParser ruleBasedSegmentParser = new RuleBasedSegmentParser();
 
         _splitFetcher = new SplitFetcherImp(splitChangeFetcher, splitParser, splitCache, _telemetryStorageProducer,
-                flagSetsFilter);
+                flagSetsFilter, ruleBasedSegmentParser, _ruleBasedSegmentCache);
 
         // SplitSynchronizationTask
         _splitSynchronizationTask = new SplitSynchronizationTask(_splitFetcher, splitCache,
@@ -617,11 +622,12 @@ private SegmentSynchronizationTaskImp buildSegments(SplitClientConfig config,
     }
 
     private SplitFetcher buildSplitFetcher(SplitCacheProducer splitCacheProducer, SplitParser splitParser,
-            FlagSetsFilter flagSetsFilter) throws URISyntaxException {
+            FlagSetsFilter flagSetsFilter, RuleBasedSegmentParser ruleBasedSegmentParser,
+           RuleBasedSegmentCacheProducer 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/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;
 }

client/src/main/java/io/split/client/utils/FeatureFlagProcessor.java

@@ -1,10 +1,14 @@
 package io.split.client.utils;
 
+import io.split.client.dtos.RuleBasedSegment;
 import io.split.client.dtos.Split;
 import io.split.client.dtos.Status;
 import io.split.client.interceptors.FlagSetsFilter;
+import io.split.engine.experiments.ParsedRuleBasedSegment;
 import io.split.engine.experiments.ParsedSplit;
+import io.split.engine.experiments.RuleBasedSegmentParser;
 import io.split.engine.experiments.SplitParser;
+import io.split.storages.RuleBasedSegmentCacheProducer;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -40,4 +44,27 @@ public static FeatureFlagsToUpdate processFeatureFlagChanges(SplitParser splitPa
         }
         return new FeatureFlagsToUpdate(toAdd, toRemove, segments);
     }
+
+    public static RuleBasedSegmentsToUpdate processRuleBasedSegmentChanges(RuleBasedSegmentParser ruleBasedSegmentParser,
+                                                                           List<RuleBasedSegment> ruleBasedSegments) {
+        List<ParsedRuleBasedSegment> toAdd = new ArrayList<>();
+        List<String> toRemove = new ArrayList<>();
+        Set<String> segments = new HashSet<>();
+        for (RuleBasedSegment ruleBasedSegment : ruleBasedSegments) {
+            if (ruleBasedSegment.status != Status.ACTIVE) {
+                // archive.
+                toRemove.add(ruleBasedSegment.name);
+                continue;
+            }
+            ParsedRuleBasedSegment parsedRuleBasedSegment = ruleBasedSegmentParser.parse(ruleBasedSegment);
+            if (parsedRuleBasedSegment == null) {
+                _log.debug(String.format("We could not parse the rule based segment definition for: %s", ruleBasedSegment.name));
+                continue;
+            }
+            segments.addAll(parsedRuleBasedSegment.getSegmentsNames());
+            toAdd.add(parsedRuleBasedSegment);
+        }
+        return new RuleBasedSegmentsToUpdate(toAdd, toRemove, segments);
+    }
+
 }

client/src/main/java/io/split/client/utils/GenericClientUtil.java

@@ -1,5 +1,10 @@
 package io.split.client.utils;
 
+import com.google.gson.JsonElement;
+import com.google.gson.JsonObject;
+import io.split.client.dtos.RuleBasedSegment;
+import io.split.client.dtos.Split;
+import io.split.client.dtos.SplitChange;
 import org.apache.hc.client5.http.classic.methods.HttpPost;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
@@ -8,6 +13,7 @@
 import org.slf4j.LoggerFactory;
 
 import java.net.URI;
+import java.util.ArrayList;
 import java.util.List;
 
 public class GenericClientUtil {
@@ -40,4 +46,25 @@ public static<T> void process(List<T> data, URI endpoint, CloseableHttpClient cl
         }
 
     }
+
+    public static SplitChange ExtractFeatureFlagsAndRuleBasedSegments(String responseBody) {
+        JsonObject jsonBody = Json.fromJson(responseBody, JsonObject.class);
+        JsonObject featureFlags = jsonBody.getAsJsonObject("ff");
+        JsonObject ruleBasedSegments = jsonBody.getAsJsonObject("rbs");
+        SplitChange splitChange = new SplitChange();
+        splitChange.till = Long.parseLong(featureFlags.get("t").toString());
+        splitChange.since = Long.parseLong(featureFlags.get("s").toString());
+        splitChange.tillRBS = Long.parseLong(ruleBasedSegments.get("t").toString());
+        splitChange.sinceRBS = Long.parseLong(ruleBasedSegments.get("s").toString());
+
+        splitChange.splits = new ArrayList<>();
+        for (JsonElement split: featureFlags.get("d").getAsJsonArray()) {
+            splitChange.splits.add(Json.fromJson(split.toString(), Split.class));
+        }
+        splitChange.ruleBasedSegments = new ArrayList<>();
+        for (JsonElement rbs: ruleBasedSegments.get("d").getAsJsonArray()) {
+            splitChange.ruleBasedSegments.add(Json.fromJson(rbs.toString(), RuleBasedSegment.class));
+        }
+        return splitChange;
+    }
 }

client/src/main/java/io/split/client/utils/RuleBasedSegmentsToUpdate.java

@@ -0,0 +1,31 @@
+package io.split.client.utils;
+
+import io.split.engine.experiments.ParsedRuleBasedSegment;
+import io.split.engine.experiments.ParsedSplit;
+
+import java.util.List;
+import java.util.Set;
+
+public class RuleBasedSegmentsToUpdate {
+    List<ParsedRuleBasedSegment> toAdd;
+    List<String> toRemove;
+    Set<String> segments;
+
+    public RuleBasedSegmentsToUpdate(List<ParsedRuleBasedSegment> toAdd, List<String> toRemove, Set<String> segments) {
+        this.toAdd = toAdd;
+        this.toRemove = toRemove;
+        this.segments = segments;
+    }
+
+    public List<ParsedRuleBasedSegment> getToAdd() {
+        return toAdd;
+    }
+
+    public List<String> getToRemove() {
+        return toRemove;
+    }
+
+    public Set<String> getSegments() {
+        return segments;
+    }
+}

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

@@ -12,6 +12,7 @@ public Builder() {}
 
         public Builder(FetchOptions opts) {
             _targetCN = opts._targetCN;
+            _targetCnRBS = opts._targetCnRBS;
             _cacheControlHeaders = opts._cacheControlHeaders;
             _flagSetsFilter = opts._flagSetsFilter;
         }
@@ -26,16 +27,22 @@ public Builder targetChangeNumber(long targetCN) {
             return this;
         }
 
+        public Builder targetChangeNumberRBS(long targetCnRBS) {
+            _targetCnRBS = targetCnRBS;
+            return this;
+        }
+
         public Builder flagSetsFilter(String flagSetsFilter) {
             _flagSetsFilter = flagSetsFilter;
             return this;
         }
 
         public FetchOptions build() {
-            return new FetchOptions(_cacheControlHeaders, _targetCN, _flagSetsFilter);
+            return new FetchOptions(_cacheControlHeaders, _targetCN, _targetCnRBS, _flagSetsFilter);
         }
 
         private long _targetCN = DEFAULT_TARGET_CHANGENUMBER;
+        private long _targetCnRBS = DEFAULT_TARGET_CHANGENUMBER;
         private boolean _cacheControlHeaders = false;
         private String _flagSetsFilter = "";
     }
@@ -46,6 +53,8 @@ public boolean cacheControlHeadersEnabled() {
 
     public long targetCN() { return _targetCN; }
 
+    public long targetCnRBS() { return _targetCnRBS; }
+
     public boolean hasCustomCN() { return _targetCN != DEFAULT_TARGET_CHANGENUMBER; }
 
     public String flagSetsFilter() {
@@ -54,9 +63,11 @@ public String flagSetsFilter() {
 
     private FetchOptions(boolean cacheControlHeaders,
                          long targetCN,
+                         long targetCnRBS,
                          String flagSetsFilter) {
         _cacheControlHeaders = cacheControlHeaders;
         _targetCN = targetCN;
+        _targetCnRBS = targetCnRBS;
         _flagSetsFilter = flagSetsFilter;
     }
 
@@ -70,6 +81,7 @@ public boolean equals(Object obj) {
 
         return Objects.equals(_cacheControlHeaders, other._cacheControlHeaders)
                 && Objects.equals(_targetCN, other._targetCN)
+                && Objects.equals(_targetCnRBS, other._targetCnRBS)
                 && Objects.equals(_flagSetsFilter, other._flagSetsFilter);
     }
 
@@ -81,5 +93,6 @@ public int hashCode() {
 
     private final boolean _cacheControlHeaders;
     private final long _targetCN;
+    private final long _targetCnRBS;
     private final String _flagSetsFilter;
 }