start cache from specific index

src/main/java/com/orbitz/consul/cache/ConsulCache.java

@@ -34,6 +34,9 @@
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.concurrent.locks.ReentrantLock;
 import java.util.function.Function;
+import java.util.function.Supplier;
+
+import javax.annotation.Nullable;
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkState;
@@ -49,7 +52,7 @@ enum State {latent, starting, started, stopped }
 
     private final static Logger LOGGER = LoggerFactory.getLogger(ConsulCache.class);
 
-    private final AtomicReference<BigInteger> latestIndex = new AtomicReference<>(null);
+    private final AtomicReference<BigInteger> latestIndex;
     private final AtomicLong lastContact = new AtomicLong();
     private final AtomicBoolean isKnownLeader = new AtomicBoolean();
     private final AtomicReference<ConsulResponse.CacheResponseInfo> lastCacheInfo = new AtomicReference<>(null);
@@ -72,9 +75,10 @@ protected ConsulCache(
             CallbackConsumer<V> callbackConsumer,
             CacheConfig cacheConfig,
             ClientEventHandler eventHandler,
-            CacheDescriptor cacheDescriptor) {
+            CacheDescriptor cacheDescriptor,
+            @Nullable BigInteger initialIndex) {
 
-        this(keyConversion, callbackConsumer, cacheConfig, eventHandler, cacheDescriptor, createDefault());
+        this(keyConversion, callbackConsumer, cacheConfig, eventHandler, cacheDescriptor, createDefault(), initialIndex);
     }
 
     protected ConsulCache(
@@ -83,9 +87,12 @@ protected ConsulCache(
             CacheConfig cacheConfig,
             ClientEventHandler eventHandler,
             CacheDescriptor cacheDescriptor,
-            ScheduledExecutorService callbackScheduleExecutorService) {
+            ScheduledExecutorService callbackScheduleExecutorService,
+            @Nullable BigInteger initialIndex) {
 
-        this(keyConversion, callbackConsumer, cacheConfig, eventHandler, cacheDescriptor, new ExternalScheduler(callbackScheduleExecutorService));
+        this(keyConversion, callbackConsumer, cacheConfig, eventHandler, cacheDescriptor,
+          new ExternalScheduler(callbackScheduleExecutorService), initialIndex
+        );
     }
 
     protected ConsulCache(
@@ -94,20 +101,14 @@ protected ConsulCache(
             CacheConfig cacheConfig,
             ClientEventHandler eventHandler,
             CacheDescriptor cacheDescriptor,
-            Scheduler callbackScheduler) {
-        if (keyConversion == null) {
-            Validate.notNull(keyConversion, "keyConversion must not be null");
-        }
-        if (callbackConsumer == null) {
-            Validate.notNull(callbackConsumer, "callbackConsumer must not be null");
-        }
-        if (cacheConfig == null) {
-            Validate.notNull(cacheConfig, "cacheConfig must not be null");
-        }
-        if (eventHandler == null) {
-            Validate.notNull(eventHandler, "eventHandler must not be null");
-        }
-
+            Scheduler callbackScheduler,
+            @Nullable BigInteger initialIndex) {
+        Validate.notNull(keyConversion, "keyConversion must not be null");
+        Validate.notNull(callbackConsumer, "callbackConsumer must not be null");
+        Validate.notNull(cacheConfig, "cacheConfig must not be null");
+        Validate.notNull(eventHandler, "eventHandler must not be null");
+
+        latestIndex = new AtomicReference<>(initialIndex);
         this.keyConversion = keyConversion;
         this.callBackConsumer = callbackConsumer;
         this.eventHandler = eventHandler;
@@ -138,28 +139,17 @@ public void onComplete(ConsulResponse<List<V>> consulResponse) {
                         // metadata changes
                         lastContact.set(consulResponse.getLastContact());
                         isKnownLeader.set(consulResponse.isKnownLeader());
-                    }
 
-                    if (changed) {
-                        Boolean locked = false;
-                        if (state.get() == State.starting) {
-                            listenersStartingLock.lock();
-                            locked = true;
-                        }
-                        try {
+                        withStartingLock(() -> {
                             for (Listener<K, V> l : listeners) {
                                 try {
                                     l.notify(full);
                                 } catch (RuntimeException e) {
                                     LOGGER.warn("ConsulCache Listener's notify method threw an exception.", e);
                                 }
                             }
-                        }
-                        finally {
-                            if (locked) {
-                                listenersStartingLock.unlock();
-                            }
-                        }
+                            return null;
+                        });
                     }
 
                     if (state.compareAndSet(State.starting, State.started)) {
@@ -173,8 +163,7 @@ public void onComplete(ConsulResponse<List<V>> consulResponse) {
                     }
                     timeToWait = timeToWait.minusMillis(elapsedTime);
 
-                    scheduler.schedule(ConsulCache.this::runCallback,
-                            timeToWait.toMillis(), TimeUnit.MILLISECONDS);
+                    scheduler.schedule(ConsulCache.this::runCallback, timeToWait.toMillis(), TimeUnit.MILLISECONDS);
 
                 } else {
                     onFailure(new ConsulException("Consul cluster has no elected leader"));
@@ -198,6 +187,21 @@ public void onFailure(Throwable throwable) {
         };
     }
 
+    private <T> T withStartingLock(Supplier<T> action) {
+      boolean wasInStartingState = state.get() == State.starting;
+      if (wasInStartingState) {
+        listenersStartingLock.lock();
+      }
+      try {
+        return action.get();
+      }
+      finally {
+        if (wasInStartingState) {
+          listenersStartingLock.unlock();
+        }
+      }
+    }
+
     static long computeBackOffDelayMs(CacheConfig cacheConfig) {
         return cacheConfig.getMinimumBackOffDelay().toMillis() +
                 Math.round(Math.random() * (cacheConfig.getMaximumBackOffDelay().minus(cacheConfig.getMinimumBackOffDelay()).toMillis()));
@@ -238,7 +242,8 @@ private void runCallback() {
     }
 
     private boolean isRunning() {
-        return state.get() == State.started || state.get() == State.starting;
+      State currentState = this.state.get();
+      return currentState == State.started || currentState == State.starting;
     }
 
     public boolean awaitInitialized(long timeout, TimeUnit unit) throws InterruptedException {
@@ -276,7 +281,8 @@ ImmutableMap<K, V> convertToMap(final ConsulResponse<List<V>> response) {
 
     private void updateIndex(ConsulResponse<List<V>> consulResponse) {
         if (consulResponse != null && consulResponse.getIndex() != null) {
-            this.latestIndex.set(consulResponse.getIndex());
+          BigInteger previousIndex = this.latestIndex.getAndSet(consulResponse.getIndex());
+          LOGGER.trace("Updated cache index from {} to {}", previousIndex, latestIndex.get());
         }
     }
 
@@ -318,6 +324,7 @@ protected static Scheduler createExternal(ScheduledExecutorService executor) {
      *
      * @param <V>
      */
+    @FunctionalInterface
     protected interface CallbackConsumer<V> {
         void consume(BigInteger index, ConsulResponseCallback<List<V>> callback);
     }
@@ -328,33 +335,23 @@ protected interface CallbackConsumer<V> {
      *
      * @param <V>
      */
+    @FunctionalInterface
     public interface Listener<K, V> {
         void notify(Map<K, V> newValues);
     }
 
     public boolean addListener(Listener<K, V> listener) {
-        Boolean locked = false;
-        boolean added;
-        if (state.get() == State.starting) {
-            listenersStartingLock.lock();
-            locked = true;
-        }
-        try {
-            added = listeners.add(listener);
-            if (state.get() == State.started) {
+        return withStartingLock(() -> {
+            boolean added = listeners.add(listener);
+            if (this.state.get() == State.started) {
                 try {
                     listener.notify(lastResponse.get());
                 } catch (RuntimeException e) {
                     LOGGER.warn("ConsulCache Listener's notify method threw an exception.", e);
                 }
             }
-        }
-        finally {
-            if (locked) {
-                listenersStartingLock.unlock();
-            }
-        }
-        return added;
+            return added;
+          });
     }
 
     public List<Listener<K, V>> getListeners() {
@@ -396,9 +393,9 @@ public DefaultScheduler() {
         }
     }
 
-    private static class ExternalScheduler extends Scheduler {
+    private static final class ExternalScheduler extends Scheduler {
 
-        public ExternalScheduler(ScheduledExecutorService executor) {
+        private ExternalScheduler(ScheduledExecutorService executor) {
             super(executor);
         }
 

src/main/java/com/orbitz/consul/cache/HealthCheckCache.java

@@ -6,6 +6,7 @@
 import com.orbitz.consul.model.health.HealthCheck;
 import com.orbitz.consul.option.QueryOptions;
 
+import java.math.BigInteger;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
 
@@ -16,7 +17,8 @@ private HealthCheckCache(HealthClient healthClient,
                              int watchSeconds,
                              QueryOptions queryOptions,
                              Function<HealthCheck, String> keyExtractor,
-                             Scheduler callbackScheduler) {
+                             Scheduler callbackScheduler,
+                             BigInteger initialIndex) {
         super(keyExtractor,
             (index, callback) -> {
                 checkWatch(healthClient.getNetworkTimeoutConfig().getClientReadTimeoutMillis(), watchSeconds);
@@ -26,7 +28,8 @@ private HealthCheckCache(HealthClient healthClient,
             healthClient.getConfig().getCacheConfig(),
             healthClient.getEventHandler(),
             new CacheDescriptor("health.state", state.getName()),
-            callbackScheduler);
+            callbackScheduler,
+            initialIndex);
     }
 
     /**
@@ -47,7 +50,20 @@ public static HealthCheckCache newCache(
             final ScheduledExecutorService callbackExecutorService) {
 
         Scheduler callbackScheduler = createExternal(callbackExecutorService);
-        return new HealthCheckCache(healthClient, state, watchSeconds, queryOptions, keyExtractor, callbackScheduler);
+        return new HealthCheckCache(healthClient, state, watchSeconds, queryOptions, keyExtractor, callbackScheduler, null);
+    }
+
+    public static HealthCheckCache newCache(
+           HealthClient healthClient,
+           com.orbitz.consul.model.State state,
+           int watchSeconds,
+           QueryOptions queryOptions,
+           BigInteger initialIndex,
+           Function<HealthCheck, String> keyExtractor,
+           ScheduledExecutorService callbackExecutorService) {
+
+      Scheduler callbackScheduler = createExternal(callbackExecutorService);
+      return new HealthCheckCache(healthClient, state, watchSeconds, queryOptions, keyExtractor, callbackScheduler, initialIndex);
     }
 
     public static HealthCheckCache newCache(
@@ -57,7 +73,7 @@ public static HealthCheckCache newCache(
             final QueryOptions queryOptions,
             final Function<HealthCheck, String> keyExtractor) {
 
-        return new HealthCheckCache(healthClient, state, watchSeconds, queryOptions, keyExtractor, createDefault());
+        return new HealthCheckCache(healthClient, state, watchSeconds, queryOptions, keyExtractor, createDefault(), null);
     }
     public static HealthCheckCache newCache(
             final HealthClient healthClient,
@@ -68,6 +84,15 @@ public static HealthCheckCache newCache(
         return newCache(healthClient, state, watchSeconds, queryOptions, HealthCheck::getCheckId);
     }
 
+    public static HealthCheckCache newCache(
+            HealthClient healthClient,
+            com.orbitz.consul.model.State state,
+            int watchSeconds,
+            QueryOptions queryOptions, BigInteger initialIndex) {
+
+      return new HealthCheckCache(healthClient, state, watchSeconds, queryOptions, HealthCheck::getCheckId, createDefault(), initialIndex);
+    }
+
     public static HealthCheckCache newCache(
             final HealthClient healthClient,
             final com.orbitz.consul.model.State state,
@@ -76,6 +101,7 @@ public static HealthCheckCache newCache(
         return newCache(healthClient, state, watchSeconds, QueryOptions.BLANK);
     }
 
+    @Deprecated
     public static HealthCheckCache newCache(final HealthClient healthClient, final com.orbitz.consul.model.State state) {
         CacheConfig cacheConfig = healthClient.getConfig().getCacheConfig();
         int watchSeconds = Ints.checkedCast(cacheConfig.getWatchDuration().getSeconds());

src/main/java/com/orbitz/consul/cache/KVCache.java

@@ -8,6 +8,7 @@
 import com.orbitz.consul.model.kv.Value;
 import com.orbitz.consul.option.QueryOptions;
 
+import java.math.BigInteger;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.function.Function;
 
@@ -18,7 +19,8 @@ private KVCache(KeyValueClient kvClient,
                     String keyPath,
                     int watchSeconds,
                     QueryOptions queryOptions,
-                    Scheduler callbackScheduler) {
+                    Scheduler callbackScheduler,
+                    BigInteger initialIndex) {
         super(getKeyExtractorFunction(keyPath),
             (index, callback) -> {
                 checkWatch(kvClient.getNetworkTimeoutConfig().getClientReadTimeoutMillis(), watchSeconds);
@@ -28,7 +30,8 @@ private KVCache(KeyValueClient kvClient,
             kvClient.getConfig().getCacheConfig(),
             kvClient.getEventHandler(),
             new CacheDescriptor("keyvalue", rootPath),
-            callbackScheduler);
+            callbackScheduler,
+            initialIndex);
     }
 
     @VisibleForTesting
@@ -56,15 +59,36 @@ public static KVCache newCache(
             final ScheduledExecutorService callbackExecutorService) {
 
         Scheduler scheduler = createExternal(callbackExecutorService);
-        return new KVCache(kvClient, rootPath, prepareRootPath(rootPath), watchSeconds, queryOptions, scheduler);
+        return new KVCache(kvClient, rootPath, prepareRootPath(rootPath), watchSeconds, queryOptions, scheduler, null);
+    }
+
+    public static KVCache newCache(
+            KeyValueClient kvClient,
+            String rootPath,
+            int watchSeconds,
+            QueryOptions queryOptions,
+            BigInteger initialIndex,
+            ScheduledExecutorService callbackExecutorService) {
+
+      Scheduler scheduler = createExternal(callbackExecutorService);
+      return new KVCache(kvClient, rootPath, prepareRootPath(rootPath), watchSeconds, queryOptions, scheduler, initialIndex);
     }
 
     public static KVCache newCache(
             final KeyValueClient kvClient,
             final String rootPath,
             final int watchSeconds,
             final QueryOptions queryOptions) {
-        return new KVCache(kvClient, rootPath, prepareRootPath(rootPath), watchSeconds, queryOptions, createDefault());
+        return new KVCache(kvClient, rootPath, prepareRootPath(rootPath), watchSeconds, queryOptions, createDefault(), null);
+    }
+
+    public static KVCache newCache(
+            KeyValueClient kvClient,
+            String rootPath,
+            int watchSeconds,
+            BigInteger initialIndex,
+            QueryOptions queryOptions) {
+      return new KVCache(kvClient, rootPath, prepareRootPath(rootPath), watchSeconds, queryOptions, createDefault(), initialIndex);
     }
 
     @VisibleForTesting
@@ -82,6 +106,7 @@ static String prepareRootPath(String rootPath) {
      *                     to be increased as well)
      * @return the cache object
      */
+    @Deprecated
     public static KVCache newCache(
             final KeyValueClient kvClient,
             final String rootPath,
@@ -97,6 +122,7 @@ public static KVCache newCache(
      * @param rootPath the root path
      * @return the cache object
      */
+    @Deprecated
     public static KVCache newCache(final KeyValueClient kvClient, final String rootPath) {
         CacheConfig cacheConfig = kvClient.getConfig().getCacheConfig();
         int watchSeconds = Ints.checkedCast(cacheConfig.getWatchDuration().getSeconds());