Merge pull request #544 from splitio/rbs-consumer

Added consumer classes

Author: chillaq

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

@@ -67,7 +67,6 @@
 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.RuleBasedSegmentCacheProducer;
 import io.split.storages.enums.OperationMode;
@@ -80,6 +79,7 @@
 import io.split.storages.pluggable.adapters.UserCustomSegmentAdapterConsumer;
 import io.split.storages.pluggable.adapters.UserCustomSplitAdapterConsumer;
 import io.split.storages.pluggable.adapters.UserCustomTelemetryAdapterProducer;
+import io.split.storages.pluggable.adapters.UserCustomRuleBasedSegmentAdapterConsumer;
 import io.split.storages.pluggable.domain.UserStorageWrapper;
 import io.split.storages.pluggable.synchronizer.TelemetryConsumerSubmitter;
 import io.split.telemetry.storage.InMemoryTelemetryStorage;
@@ -341,8 +341,8 @@ protected SplitFactoryImpl(String apiToken, SplitClientConfig config, CustomStor
         _gates = new SDKReadinessGates();
 
         _telemetrySynchronizer = new TelemetryConsumerSubmitter(customStorageWrapper, _sdkMetadata);
-        // TODO Update the instance to UserCustomRuleBasedSegmentAdapterConsumer
-        RuleBasedSegmentCacheConsumer userCustomRuleBasedSegmentAdapterConsumer = new RuleBasedSegmentCacheInMemoryImp();
+        UserCustomRuleBasedSegmentAdapterConsumer userCustomRuleBasedSegmentAdapterConsumer =
+                new UserCustomRuleBasedSegmentAdapterConsumer(customStorageWrapper);
         _evaluator = new EvaluatorImp(userCustomSplitAdapterConsumer, userCustomSegmentAdapterConsumer, userCustomRuleBasedSegmentAdapterConsumer);
         _impressionsSender = PluggableImpressionSender.create(customStorageWrapper);
         _uniqueKeysTracker = createUniqueKeysTracker(config);

client/src/main/java/io/split/storages/pluggable/adapters/UserCustomRuleBasedSegmentAdapterConsumer.java

@@ -0,0 +1,95 @@
+package io.split.storages.pluggable.adapters;
+
+import io.split.client.dtos.RuleBasedSegment;
+import io.split.client.utils.Json;
+import io.split.engine.experiments.ParsedRuleBasedSegment;
+import io.split.engine.experiments.RuleBasedSegmentParser;
+import io.split.storages.RuleBasedSegmentCacheConsumer;
+import io.split.storages.pluggable.domain.PrefixAdapter;
+import io.split.storages.pluggable.domain.UserStorageWrapper;
+import io.split.storages.pluggable.utils.Helper;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import pluggable.CustomStorageWrapper;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.stream.Collectors;
+
+import static com.google.common.base.Preconditions.checkNotNull;
+
+public class UserCustomRuleBasedSegmentAdapterConsumer implements RuleBasedSegmentCacheConsumer {
+
+    private static final Logger _log = LoggerFactory.getLogger(UserCustomRuleBasedSegmentAdapterConsumer.class);
+
+    private final RuleBasedSegmentParser _ruleBasedSegmentParser;
+    private final UserStorageWrapper _userStorageWrapper;
+
+    public UserCustomRuleBasedSegmentAdapterConsumer(CustomStorageWrapper customStorageWrapper) {
+        _ruleBasedSegmentParser = new RuleBasedSegmentParser();
+        _userStorageWrapper = new UserStorageWrapper(checkNotNull(customStorageWrapper));
+    }
+
+    @Override
+    public long getChangeNumber() {
+        String wrapperResponse = _userStorageWrapper.get(PrefixAdapter.buildRuleBasedSegmentChangeNumber());
+        return Helper.responseToLong(wrapperResponse, -1L);
+    }
+
+    @Override
+    public ParsedRuleBasedSegment get(String name) {
+        String wrapperResponse = _userStorageWrapper.get(PrefixAdapter.buildRuleBasedSegmentKey(name));
+        if(wrapperResponse == null) {
+            return null;
+        }
+        RuleBasedSegment ruleBasedSegment = Json.fromJson(wrapperResponse, RuleBasedSegment.class);
+        if(ruleBasedSegment == null) {
+            _log.warn("Could not parse RuleBasedSegment.");
+            return null;
+        }
+        return _ruleBasedSegmentParser.parse(ruleBasedSegment);
+    }
+
+    @Override
+    public Collection<ParsedRuleBasedSegment> getAll() {
+        Set<String> keys = _userStorageWrapper.getKeysByPrefix(PrefixAdapter.buildGetAllRuleBasedSegment());
+        if(keys == null) {
+            return new ArrayList<>();
+        }
+        List<String> wrapperResponse = _userStorageWrapper.getMany(new ArrayList<>(keys));
+        if(wrapperResponse == null) {
+            return new ArrayList<>();
+        }
+        return stringsToParsedRuleBasedSegments(wrapperResponse);
+    }
+
+    @Override
+    public List<String> ruleBasedSegmentNames() {
+        Set<String> ruleBasedSegmentNamesWithPrefix = _userStorageWrapper.getKeysByPrefix(PrefixAdapter.buildGetAllRuleBasedSegment());
+        ruleBasedSegmentNamesWithPrefix = ruleBasedSegmentNamesWithPrefix.stream().
+                map(key -> key.replace(PrefixAdapter.buildRuleBasedSegmentsPrefix(), "")).
+                collect(Collectors.toSet());
+        return new ArrayList<>(ruleBasedSegmentNamesWithPrefix);
+    }
+
+    @Override
+    public Set<String> getSegments() {
+        return getAll().stream()
+                .flatMap(parsedRuleBasedSegment -> parsedRuleBasedSegment.
+                        getSegmentsNames().stream()).collect(Collectors.toSet());
+    }
+
+    private List<ParsedRuleBasedSegment> stringsToParsedRuleBasedSegments(List<String> elements) {
+        List<ParsedRuleBasedSegment> result = new ArrayList<>();
+        for(String s : elements) {
+            if(s != null) {
+                result.add(_ruleBasedSegmentParser.parse(Json.fromJson(s, RuleBasedSegment.class)));
+                continue;
+            }
+            result.add(null);
+        }
+        return result;
+    }
+}

client/src/main/java/io/split/storages/pluggable/domain/PrefixAdapter.java

@@ -20,6 +20,8 @@ public class PrefixAdapter {
     private static final String EXCEPTIONS = "exceptions";
     private static final String INIT = "init";
     private static final String FLAG_SET = "flagSet";
+    private static final String RULE_BASED_SEGMENT_PREFIX = "rbsegment";
+    private static final String RULE_BASED_SEGMENTS_PREFIX = "rbsegments";
 
     public static String buildSplitKey(String name) {
         return String.format(DEFAULT_PREFIX+ SPLIT_PREFIX +"%s", name);
@@ -37,6 +39,22 @@ public static String buildSplitsPrefix(){
         return DEFAULT_PREFIX+SPLIT_PREFIX;
     }
 
+    public static String buildRuleBasedSegmentKey(String name) {
+        return String.format(DEFAULT_PREFIX+ RULE_BASED_SEGMENT_PREFIX +"%s", name);
+    }
+
+    public static String buildRuleBasedSegmentsPrefix(){
+        return DEFAULT_PREFIX+RULE_BASED_SEGMENT_PREFIX;
+    }
+
+    public static String buildRuleBasedSegmentChangeNumber() {
+        return DEFAULT_PREFIX+RULE_BASED_SEGMENTS_PREFIX+"till";
+    }
+
+    public static String buildGetAllRuleBasedSegment() {
+        return DEFAULT_PREFIX+RULE_BASED_SEGMENT_PREFIX+"*";
+    }
+
     public static String buildTrafficTypeExists(String trafficType) {
         return String.format(DEFAULT_PREFIX+TRAFFIC_TYPE_PREFIX+"%s", trafficType);
     }

client/src/test/java/io/split/TestHelper.java

@@ -1,5 +1,9 @@
 package io.split;
 
+import io.split.client.dtos.Condition;
+import io.split.client.dtos.Excluded;
+import io.split.client.dtos.RuleBasedSegment;
+import io.split.client.dtos.Status;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpClient;
 import org.apache.hc.client5.http.impl.classic.CloseableHttpResponse;
 import org.apache.hc.core5.http.ClassicHttpResponse;
@@ -12,6 +16,8 @@
 import java.io.IOException;
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
+import java.util.ArrayList;
+import java.util.List;
 
 public class TestHelper {
     public static CloseableHttpClient mockHttpClient(String jsonName, int httpStatus) throws IOException, IllegalAccessException, NoSuchMethodException, InvocationTargetException {
@@ -36,4 +42,20 @@ public static CloseableHttpResponse classicResponseToCloseableMock(ClassicHttpRe
         adaptMethod.setAccessible(true);
         return (CloseableHttpResponse) adaptMethod.invoke(null, mocked);
     }
+
+    public static RuleBasedSegment makeRuleBasedSegment(String name, List<Condition> conditions, long changeNumber) {
+        Excluded excluded = new Excluded();
+        excluded.segments = new ArrayList<>();
+        excluded.keys = new ArrayList<>();
+
+        RuleBasedSegment ruleBasedSegment = new RuleBasedSegment();
+        ruleBasedSegment.name = name;
+        ruleBasedSegment.status = Status.ACTIVE;
+        ruleBasedSegment.conditions = conditions;
+        ruleBasedSegment.trafficTypeName = "user";
+        ruleBasedSegment.changeNumber = changeNumber;
+        ruleBasedSegment.excluded = excluded;
+        return ruleBasedSegment;
+    }
+
 }

client/src/test/java/io/split/engine/experiments/RuleBasedSegmentParserTest.java

@@ -31,7 +31,9 @@
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
+import static io.split.TestHelper.makeRuleBasedSegment;
 import static io.split.client.utils.RuleBasedSegmentProcessor.processRuleBasedSegmentChanges;
+
 import static org.junit.Assert.assertTrue;
 
 /**
@@ -542,21 +544,6 @@ public void setMatcherTest(Condition c, io.split.engine.matchers.Matcher m) {
         Assert.assertEquals(actual, expected);
     }
 
-    private RuleBasedSegment makeRuleBasedSegment(String name, List<Condition> conditions, long changeNumber) {
-        Excluded excluded = new Excluded();
-        excluded.segments = new ArrayList<>();
-        excluded.keys = new ArrayList<>();
-
-        RuleBasedSegment ruleBasedSegment = new RuleBasedSegment();
-        ruleBasedSegment.name = name;
-        ruleBasedSegment.status = Status.ACTIVE;
-        ruleBasedSegment.conditions = conditions;
-        ruleBasedSegment.trafficTypeName = "user";
-        ruleBasedSegment.changeNumber = changeNumber;
-        ruleBasedSegment.excluded = excluded;
-        return ruleBasedSegment;
-    }
-
     private SegmentChange getSegmentChange(long since, long till, String segmentName){
         SegmentChange segmentChange = new SegmentChange();
         segmentChange.name = segmentName;