Merge pull request #175 from splitio/split-fetcher-refactor-storage

Splits fetcher: Decouple storage

Author: sanzmauro

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

@@ -192,7 +192,7 @@ private boolean track(Event event) {
             event.trafficTypeName = event.trafficTypeName.toLowerCase();
         }
 
-        if (!_splitFetcher.fetchKnownTrafficTypes().contains(event.trafficTypeName)) {
+        if (!_splitFetcher.trafficTypeExists(event.trafficTypeName)) {
             _log.warn("track: Traffic Type " + event.trafficTypeName + " does not have any corresponding Splits in this environment, " +
                     "make sure you’re tracking your events to a valid traffic type defined in the Split console.");
         }

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

@@ -11,6 +11,8 @@
 import io.split.client.metrics.CachedMetrics;
 import io.split.client.metrics.FireAndForgetMetrics;
 import io.split.client.metrics.HttpMetrics;
+import io.split.engine.cache.InMemoryCacheImp;
+import io.split.engine.cache.SplitCache;
 import io.split.engine.evaluator.Evaluator;
 import io.split.engine.evaluator.EvaluatorImp;
 import io.split.engine.SDKReadinessGates;
@@ -200,7 +202,8 @@ public SplitFactoryImpl(String apiToken, SplitClientConfig config) throws URISyn
         // Feature Changes
         SplitChangeFetcher splitChangeFetcher = HttpSplitChangeFetcher.create(httpclient, rootTarget, uncachedFireAndForget);
 
-        final RefreshableSplitFetcherProvider splitFetcherProvider = new RefreshableSplitFetcherProvider(splitChangeFetcher, splitParser, findPollingPeriod(RANDOM, config.featuresRefreshRate()), gates);
+        final SplitCache cache = new InMemoryCacheImp();
+        final RefreshableSplitFetcherProvider splitFetcherProvider = new RefreshableSplitFetcherProvider(splitChangeFetcher, splitParser, findPollingPeriod(RANDOM, config.featuresRefreshRate()), gates, cache);
 
 
         List<ImpressionListener> impressionListeners = new ArrayList<>();

client/src/main/java/io/split/engine/cache/InMemoryCacheImp.java

@@ -0,0 +1,108 @@
+package io.split.engine.cache;
+
+import com.google.common.collect.ConcurrentHashMultiset;
+import com.google.common.collect.Maps;
+import com.google.common.collect.Multiset;
+import com.google.common.collect.Sets;
+import io.split.engine.experiments.ParsedSplit;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.util.ArrayList;
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+import java.util.concurrent.ConcurrentMap;
+import java.util.concurrent.atomic.AtomicLong;
+
+public class InMemoryCacheImp implements SplitCache {
+
+    private static final Logger _log = LoggerFactory.getLogger(InMemoryCacheImp.class);
+
+    private final ConcurrentMap<String, ParsedSplit> _concurrentMap;
+    private final Multiset<String> _concurrentTrafficTypeNameSet;
+
+    private AtomicLong _changeNumber;
+
+    public InMemoryCacheImp() {
+        this(-1);
+    }
+
+    public InMemoryCacheImp(long startingChangeNumber) {
+        _concurrentMap = Maps.newConcurrentMap();
+        _changeNumber = new AtomicLong(startingChangeNumber);
+        _concurrentTrafficTypeNameSet = ConcurrentHashMultiset.create();
+    }
+
+    @Override
+    public void put(ParsedSplit split) {
+        _concurrentMap.put(split.feature(), split);
+
+        if (split.trafficTypeName() != null) {
+            _concurrentTrafficTypeNameSet.add(split.trafficTypeName());
+        }
+    }
+
+    @Override
+    public boolean remove(String name) {
+        ParsedSplit removed = _concurrentMap.remove(name);
+
+        if (removed != null && removed.trafficTypeName() != null) {
+            _concurrentTrafficTypeNameSet.remove(removed.trafficTypeName());
+        }
+
+        return removed != null;
+    }
+
+    @Override
+    public ParsedSplit get(String name) {
+        return _concurrentMap.get(name);
+    }
+
+    @Override
+    public Collection<ParsedSplit> getAll() {
+        return _concurrentMap.values();
+    }
+
+    @Override
+    public Collection<ParsedSplit> getMany(List<String> names) {
+        List<ParsedSplit> splits = new ArrayList<>();
+
+        for (String name : names) {
+            ParsedSplit split = _concurrentMap.get(name);
+
+            if (split != null) {
+                splits.add(split);
+            }
+        }
+
+        return splits;
+    }
+
+    @Override
+    public long getChangeNumber() {
+        return _changeNumber.get();
+    }
+
+    @Override
+    public void setChangeNumber(long changeNumber) {
+        if (changeNumber < _changeNumber.get()) {
+            _log.error("ChangeNumber for splits cache is less than previous");
+        }
+
+        _changeNumber.set(changeNumber);
+    }
+
+    @Override
+    public boolean trafficTypeExists(String trafficTypeName) {
+        // If the multiset has [{"user",2}.{"account",0}], elementSet only returns
+        // ["user"] (it ignores "account")
+        return Sets.newHashSet(_concurrentTrafficTypeNameSet.elementSet()).contains(trafficTypeName);
+    }
+
+    @Override
+    public void clear() {
+        _concurrentMap.clear();
+        _concurrentTrafficTypeNameSet.clear();
+    }
+}

client/src/main/java/io/split/engine/cache/SplitCache.java

@@ -0,0 +1,19 @@
+package io.split.engine.cache;
+
+import io.split.engine.experiments.ParsedSplit;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Set;
+
+public interface SplitCache {
+    void put(ParsedSplit split);
+    boolean remove(String name);
+    ParsedSplit get(String name);
+    Collection<ParsedSplit> getAll();
+    Collection<ParsedSplit> getMany(List<String> names);
+    long getChangeNumber();
+    void setChangeNumber(long changeNumber);
+    boolean trafficTypeExists(String trafficTypeName);
+    void clear();
+}