--- /dev/null
+package org.onap.msb.apiroute.wrapper.consulextend.cache;
+
+import static com.google.common.base.Preconditions.checkArgument;
+import static com.google.common.base.Preconditions.checkState;
+
+import java.math.BigInteger;
+import java.util.Collections;
+import java.util.List;
+import java.util.Properties;
+import java.util.concurrent.CopyOnWriteArrayList;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executors;
+import java.util.concurrent.ScheduledExecutorService;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import org.onap.msb.apiroute.wrapper.consulextend.async.ConsulResponseCallback;
+import org.onap.msb.apiroute.wrapper.consulextend.async.ConsulResponseHeader;
+import org.onap.msb.apiroute.wrapper.consulextend.async.OriginalConsulResponse;
+import org.onap.msb.apiroute.wrapper.consulextend.util.Http;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import com.google.common.annotations.VisibleForTesting;
+import com.google.common.base.Strings;
+import com.orbitz.consul.ConsulException;
+import com.orbitz.consul.model.ConsulResponse;
+import com.orbitz.consul.option.ImmutableQueryOptions;
+import com.orbitz.consul.option.QueryOptions;
+
+/**
+ * A cache structure that can provide an up-to-date read-only map backed by
+ * consul data
+ *
+ * @param <V>
+ */
+public class ConsulCache<T> {
+
+ enum State {
+ latent, starting, started, stopped
+ }
+
+ private final static Logger LOGGER = LoggerFactory
+ .getLogger(ConsulCache.class);
+
+ @VisibleForTesting
+ static final String BACKOFF_DELAY_PROPERTY = "com.orbitz.consul.cache.backOffDelay";
+ private static final long BACKOFF_DELAY_QTY_IN_MS = getBackOffDelayInMs(System
+ .getProperties());
+
+ private final AtomicReference<BigInteger> latestIndex = new AtomicReference<BigInteger>(
+ null);
+ private final AtomicReference<State> state = new AtomicReference<State>(
+ State.latent);
+ private final CountDownLatch initLatch = new CountDownLatch(1);
+ private final ScheduledExecutorService executorService = Executors
+ .newSingleThreadScheduledExecutor();
+ private final CopyOnWriteArrayList<Listener<T>> listeners = new CopyOnWriteArrayList<Listener<T>>();
+
+ private final CallbackConsumer<T> callBackConsumer;
+ private final ConsulResponseCallback<T> responseCallback;
+
+ ConsulCache(CallbackConsumer<T> callbackConsumer) {
+
+ this.callBackConsumer = callbackConsumer;
+
+ this.responseCallback = new ConsulResponseCallback<T>() {
+ @Override
+ public void onComplete(ConsulResponse<T> consulResponse) {
+
+ if (consulResponse.isKnownLeader()) {
+ if (!isRunning()) {
+ return;
+ }
+ updateIndex(consulResponse);
+
+ for (Listener<T> l : listeners) {
+ l.notify(consulResponse);
+ }
+
+ if (state.compareAndSet(State.starting, State.started)) {
+ initLatch.countDown();
+ }
+
+ runCallback();
+ } else {
+ onFailure(new ConsulException(
+ "Consul cluster has no elected leader"));
+ }
+ }
+
+ @Override
+ public void onDelayComplete(
+ OriginalConsulResponse<T> originalConsulResponse) {
+
+ try {
+ // get header
+ ConsulResponseHeader consulResponseHeader = Http
+ .consulResponseHeader(originalConsulResponse
+ .getResponse());
+
+ if (consulResponseHeader.isKnownLeader()) {
+ if (!isRunning()) {
+ return;
+ }
+
+ boolean isConuslIndexChanged = isConuslIndexChanged(consulResponseHeader
+ .getIndex());
+ // consul index different
+ if (isConuslIndexChanged) {
+
+ updateIndex(consulResponseHeader.getIndex());
+
+ // get T type data
+ ConsulResponse<T> consulResponse = Http
+ .consulResponse(originalConsulResponse
+ .getResponseType(),
+ originalConsulResponse
+ .getResponse());
+
+ // notify customer to custom T data
+ for (Listener<T> l : listeners) {
+ l.notify(consulResponse);
+ }
+ }
+
+ if (state.compareAndSet(State.starting, State.started)) {
+ initLatch.countDown();
+ }
+
+ runCallback();
+
+ } else {
+ onFailure(new ConsulException(
+ "Consul cluster has no elected leader"));
+ }
+ } catch (Exception e) {
+ onFailure(e);
+ }
+
+ }
+
+ @Override
+ public void onFailure(Throwable throwable) {
+
+ if (!isRunning()) {
+ return;
+ }
+ LOGGER.error(
+ String.format(
+ "Error getting response from consul. will retry in %d %s",
+ BACKOFF_DELAY_QTY_IN_MS, TimeUnit.MILLISECONDS),
+ throwable);
+
+ executorService.schedule(new Runnable() {
+ @Override
+ public void run() {
+ runCallback();
+ }
+ }, BACKOFF_DELAY_QTY_IN_MS, TimeUnit.MILLISECONDS);
+ }
+ };
+ }
+
+ @VisibleForTesting
+ static long getBackOffDelayInMs(Properties properties) {
+ String backOffDelay = null;
+ try {
+ backOffDelay = properties.getProperty(BACKOFF_DELAY_PROPERTY);
+ if (!Strings.isNullOrEmpty(backOffDelay)) {
+ return Long.parseLong(backOffDelay);
+ }
+ } catch (Exception ex) {
+ LOGGER.warn(
+ backOffDelay != null ? String.format(
+ "Error parsing property variable %s: %s",
+ BACKOFF_DELAY_PROPERTY, backOffDelay) : String
+ .format("Error extracting property variable %s",
+ BACKOFF_DELAY_PROPERTY), ex);
+ }
+ return TimeUnit.SECONDS.toMillis(10);
+ }
+
+ public void start() throws Exception {
+ checkState(state.compareAndSet(State.latent, State.starting),
+ "Cannot transition from state %s to %s", state.get(),
+ State.starting);
+ runCallback();
+ }
+
+ public void stop() throws Exception {
+ State previous = state.getAndSet(State.stopped);
+ if (previous != State.stopped) {
+ executorService.shutdownNow();
+ }
+ }
+
+ private void runCallback() {
+ if (isRunning()) {
+ callBackConsumer.consume(latestIndex.get(), responseCallback);
+ }
+ }
+
+ private boolean isRunning() {
+ return state.get() == State.started || state.get() == State.starting;
+ }
+
+ public boolean awaitInitialized(long timeout, TimeUnit unit)
+ throws InterruptedException {
+ return initLatch.await(timeout, unit);
+ }
+
+ private void updateIndex(ConsulResponse<T> consulResponse) {
+ if (consulResponse != null && consulResponse.getIndex() != null) {
+ this.latestIndex.set(consulResponse.getIndex());
+ }
+ }
+
+ public void updateIndex(BigInteger index) {
+ if (index != null) {
+ this.latestIndex.set(index);
+ }
+ }
+
+ protected static QueryOptions watchParams(final BigInteger index,
+ final int blockSeconds, QueryOptions queryOptions) {
+ checkArgument(!queryOptions.getIndex().isPresent()
+ && !queryOptions.getWait().isPresent(),
+ "Index and wait cannot be overridden");
+
+ return ImmutableQueryOptions.builder()
+ .from(watchDefaultParams(index, blockSeconds))
+ .token(queryOptions.getToken())
+ .consistencyMode(queryOptions.getConsistencyMode())
+ .near(queryOptions.getNear()).build();
+ }
+
+ private static QueryOptions watchDefaultParams(final BigInteger index,
+ final int blockSeconds) {
+ if (index == null) {
+ return QueryOptions.BLANK;
+ } else {
+ return QueryOptions.blockSeconds(blockSeconds, index).build();
+ }
+ }
+
+ /**
+ * passed in by creators to vary the content of the cached values
+ *
+ * @param <V>
+ */
+ protected interface CallbackConsumer<T> {
+ void consume(BigInteger index, ConsulResponseCallback<T> callback);
+ }
+
+ /**
+ * Implementers can register a listener to receive a new map when it changes
+ *
+ * @param <V>
+ */
+ public interface Listener<T> {
+ void notify(ConsulResponse<T> newValues);
+ }
+
+ public boolean addListener(Listener<T> listener) {
+ boolean added = listeners.add(listener);
+ return added;
+ }
+
+ public List<Listener<T>> getListeners() {
+ return Collections.unmodifiableList(listeners);
+ }
+
+ public boolean removeListener(Listener<T> listener) {
+ return listeners.remove(listener);
+ }
+
+ @VisibleForTesting
+ protected State getState() {
+ return state.get();
+ }
+
+ private boolean isConuslIndexChanged(final BigInteger index) {
+
+ if (index != null && !index.equals(latestIndex.get())) {
+
+ if (LOGGER.isDebugEnabled()) {
+ // 第一次不打印
+ if (latestIndex.get() != null) {
+ LOGGER.debug("consul index compare:new-" + index + " old-"
+ + latestIndex.get());
+ }
+
+ }
+
+ return true;
+ }
+
+ return false;
+ }
+}
Code Review / msb / apigateway.git / blobdiff