added splitCache and splitFetcher refactor

Author: sanzmauro

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(-1);
+        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,92 @@
+package io.split.engine.cache;
+
+import com.google.common.collect.Maps;
+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.Map;
+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 AtomicLong _changeNumber;
+
+    public InMemoryCacheImp(long startingChangeNumber) {
+        _concurrentMap = Maps.newConcurrentMap();
+        _changeNumber = new AtomicLong(startingChangeNumber);
+    }
+
+    @Override
+    public void put(ParsedSplit split) {
+        _concurrentMap.put(split.feature(), split);
+    }
+
+    @Override
+    public void putAll(Map<String, ParsedSplit> splits) {
+        _concurrentMap.putAll(splits);
+    }
+
+    @Override
+    public boolean remove(String name) {
+        ParsedSplit removed = _concurrentMap.remove(name);
+
+        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 trafficType) {
+        return false;
+    }
+
+    @Override
+    public void clear() {
+        _concurrentMap.clear();
+    }
+}

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

@@ -0,0 +1,20 @@
+package io.split.engine.cache;
+
+import io.split.engine.experiments.ParsedSplit;
+
+import java.util.Collection;
+import java.util.List;
+import java.util.Map;
+
+public interface SplitCache {
+    void put(ParsedSplit split);
+    void putAll(Map<String, ParsedSplit> splits);
+    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 trafficType);
+    void clear();
+}

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

@@ -6,21 +6,18 @@
 import com.google.common.collect.Multiset;
 import com.google.common.collect.Multisets;
 import com.google.common.collect.Sets;
-import io.split.client.dtos.Condition;
-import io.split.client.dtos.Matcher;
-import io.split.client.dtos.MatcherType;
 import io.split.client.dtos.Split;
 import io.split.client.dtos.SplitChange;
 import io.split.client.dtos.Status;
 import io.split.engine.SDKReadinessGates;
+import io.split.engine.cache.SplitCache;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 import java.util.Collection;
 import java.util.List;
 import java.util.Map;
 import java.util.Set;
-import java.util.concurrent.atomic.AtomicLong;
 
 import static com.google.common.base.Preconditions.checkNotNull;
 
@@ -35,9 +32,9 @@ public class RefreshableSplitFetcher implements SplitFetcher, Runnable {
 
     private final SplitParser _parser;
     private final SplitChangeFetcher _splitChangeFetcher;
-    private final AtomicLong _changeNumber;
-
-    private Map<String, ParsedSplit> _concurrentMap = Maps.newConcurrentMap();
+    private final SplitCache _splitCache;
+    private final SDKReadinessGates _gates;
+    private final Object _lock = new Object();
 
     /**
      * Contains all the traffic types that are currently being used by the splits and also the count
@@ -49,45 +46,22 @@ public class RefreshableSplitFetcher implements SplitFetcher, Runnable {
      * an ARCHIVED split is received, we know if we need to remove a traffic type from the multiset.
      */
     Multiset<String> _concurrentTrafficTypeNameSet = ConcurrentHashMultiset.create();
-    private final SDKReadinessGates _gates;
-
-    private final Object _lock = new Object();
 
-
-    public RefreshableSplitFetcher(SplitChangeFetcher splitChangeFetcher, SplitParser parser, SDKReadinessGates gates) {
-        this(splitChangeFetcher, parser, gates, -1);
-    }
-
-    /**
-     * This constructor is package private because it is meant primarily for unit tests
-     * where we want to set the starting change number. All regular clients should use
-     * the public constructor.
-     *
-     * @param splitChangeFetcher   MUST NOT be null
-     * @param parser               MUST NOT be null
-     * @param startingChangeNumber
-     */
-    /*package private*/ RefreshableSplitFetcher(SplitChangeFetcher splitChangeFetcher,
-                                                SplitParser parser,
-                                                SDKReadinessGates gates,
-                                                long startingChangeNumber) {
-        _splitChangeFetcher = splitChangeFetcher;
-        _parser = parser;
-        _gates = gates;
-        _changeNumber = new AtomicLong(startingChangeNumber);
-
-        checkNotNull(_parser);
-        checkNotNull(_splitChangeFetcher);
+    public RefreshableSplitFetcher(SplitChangeFetcher splitChangeFetcher, SplitParser parser, SDKReadinessGates gates, SplitCache splitCache) {
+        _splitChangeFetcher = checkNotNull(splitChangeFetcher);
+        _parser = checkNotNull(parser);
+        _gates = checkNotNull(gates);
+        _splitCache = checkNotNull(splitCache);
     }
 
     @Override
     public void forceRefresh() {
         _log.debug("Force Refresh splits starting ...");
         try {
             while (true) {
-                long start = _changeNumber.get();
+                long start = _splitCache.getChangeNumber();
                 runWithoutExceptionHandling();
-                long end = _changeNumber.get();
+                long end = _splitCache.getChangeNumber();
 
                 if (start >= end) {
                     break;
@@ -103,13 +77,13 @@ public void forceRefresh() {
 
     @Override
     public long changeNumber() {
-        return _changeNumber.get();
+        return _splitCache.getChangeNumber();
     }
 
     @Override
     public void killSplit(String splitName, String defaultTreatment, long changeNumber) {
         synchronized (_lock) {
-            ParsedSplit parsedSplit = _concurrentMap.get(splitName);
+            ParsedSplit parsedSplit = _splitCache.get(splitName);
 
             ParsedSplit updatedSplit = new ParsedSplit(parsedSplit.feature(),
                     parsedSplit.seed(),
@@ -123,17 +97,17 @@ public void killSplit(String splitName, String defaultTreatment, long changeNumb
                     parsedSplit.algo(),
                     parsedSplit.configurations());
 
-            _concurrentMap.put(splitName, updatedSplit);
+            _splitCache.put(updatedSplit);
         }
     }
 
     @Override
     public ParsedSplit fetch(String test) {
-        return _concurrentMap.get(test);
+        return _splitCache.get(test);
     }
 
     public List<ParsedSplit> fetchAll() {
-        return Lists.newArrayList(_concurrentMap.values());
+        return Lists.newArrayList(_splitCache.getAll());
     }
 
     @Override
@@ -145,18 +119,18 @@ public Set<String> fetchKnownTrafficTypes() {
     }
 
     public Collection<ParsedSplit> fetch() {
-        return _concurrentMap.values();
+        return _splitCache.getAll();
     }
 
     public void clear() {
-        _concurrentMap.clear();
+        _splitCache.clear();
         _concurrentTrafficTypeNameSet.clear();
     }
 
     @Override
     public void run() {
         _log.debug("Fetch splits starting ...");
-        long start = _changeNumber.get();
+        long start = _splitCache.getChangeNumber();
         try {
             runWithoutExceptionHandling();
             _gates.splitsAreReady();
@@ -170,38 +144,38 @@ public void run() {
             }
         } finally {
             if (_log.isDebugEnabled()) {
-                _log.debug("split fetch before: " + start + ", after: " + _changeNumber.get());
+                _log.debug("split fetch before: " + start + ", after: " + _splitCache.getChangeNumber());
             }
         }
     }
 
     public void runWithoutExceptionHandling() throws InterruptedException {
-        SplitChange change = _splitChangeFetcher.fetch(_changeNumber.get());
+        SplitChange change = _splitChangeFetcher.fetch(_splitCache.getChangeNumber());
 
         if (change == null) {
             throw new IllegalStateException("SplitChange was null");
         }
 
-        if (change.till == _changeNumber.get()) {
+        if (change.till == _splitCache.getChangeNumber()) {
             // no change.
             return;
         }
 
-        if (change.since != _changeNumber.get() || change.till < _changeNumber.get()) {
+        if (change.since != _splitCache.getChangeNumber() || change.till < _splitCache.getChangeNumber()) {
             // some other thread may have updated the shared state. exit
             return;
         }
 
         if (change.splits.isEmpty()) {
             // there are no changes. weird!
-            _changeNumber.set(change.till);
+            _splitCache.setChangeNumber(change.till);
             return;
         }
 
         synchronized (_lock) {
             // check state one more time.
-            if (change.since != _changeNumber.get()
-                    || change.till < _changeNumber.get()) {
+            if (change.since != _splitCache.getChangeNumber()
+                    || change.till < _splitCache.getChangeNumber()) {
                 // some other thread may have updated the shared state. exit
                 return;
             }
@@ -243,7 +217,7 @@ public void runWithoutExceptionHandling() throws InterruptedException {
                 // If it's deleted & recreated, the old one should be decreased and the new one increased.
                 // To handle both cases, we simply delete the old one if the split is present.
                 // The new one is always increased.
-                ParsedSplit current = _concurrentMap.get(split.name);
+                ParsedSplit current = _splitCache.get(split.name);
                 if (current != null && current.trafficTypeName() != null) {
                     trafficTypeNamesToRemove.add(current.trafficTypeName());
                 }
@@ -253,13 +227,13 @@ public void runWithoutExceptionHandling() throws InterruptedException {
                 }
             }
 
-            _concurrentMap.putAll(toAdd);
+            _splitCache.putAll(toAdd);
             _concurrentTrafficTypeNameSet.addAll(trafficTypeNamesToAdd);
             //removeAll does not work here, since it wont remove all the occurrences, just one
             Multisets.removeOccurrences(_concurrentTrafficTypeNameSet, trafficTypeNamesToRemove);
 
             for (String remove : toRemove) {
-                _concurrentMap.remove(remove);
+                _splitCache.remove(remove);
             }
 
             if (!toAdd.isEmpty()) {
@@ -270,7 +244,7 @@ public void runWithoutExceptionHandling() throws InterruptedException {
                 _log.debug("Deleted features: " + toRemove);
             }
 
-            _changeNumber.set(change.till);
+            _splitCache.setChangeNumber(change.till);
         }
     }
 }