opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/SequencedQueueEntry.java

   1 /*
   2  * Copyright (c) 2016 Cisco Systems, Inc. and others.  All rights reserved.
   3  *
   4  * This program and the accompanying materials are made available under the
   5  * terms of the Eclipse Public License v1.0 which accompanies this distribution,
   6  * and is available at http://www.eclipse.org/legal/epl-v10.html
   7  */
   8 package org.opendaylight.controller.cluster.access.client;
   9
  10 import akka.actor.ActorRef;
  11 import com.google.common.base.MoreObjects;
  12 import com.google.common.base.Preconditions;
  13 import java.util.Optional;
  14 import org.opendaylight.controller.cluster.access.concepts.Request;
  15 import org.opendaylight.controller.cluster.access.concepts.RequestEnvelope;
  16 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  17 import org.opendaylight.controller.cluster.access.concepts.Response;
  18 import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
  19 import org.slf4j.Logger;
  20 import org.slf4j.LoggerFactory;
  21
  22 /**
  23  * Single entry in {@link SequencedQueue}. Tracks the request, the associated callback and accounting information.
  24  *
  25  * @author Robert Varga
  26  *
  27  * @param <I> Target identifier type
  28  */
  29 final class SequencedQueueEntry {
  30     private static final class LastTry {
  31         final long timeTicks;
  32         final long retry;
  33
  34         LastTry(final long retry, final long timeTicks) {
  35             this.retry = retry;
  36             this.timeTicks = timeTicks;
  37         }
  38     }
  39
  40     private static final Logger LOG = LoggerFactory.getLogger(SequencedQueueEntry.class);
  41
  42     private final Request<?, ?> request;
  43     private final RequestCallback callback;
  44     private final long enqueuedTicks;
  45     private final long sequence;
  46
  47     private Optional<LastTry> lastTry = Optional.empty();
  48
  49     SequencedQueueEntry(final Request<?, ?> request, final long sequence, final RequestCallback callback,
  50         final long now) {
  51         this.request = Preconditions.checkNotNull(request);
  52         this.callback = Preconditions.checkNotNull(callback);
  53         this.enqueuedTicks = now;
  54         this.sequence = sequence;
  55     }
  56
  57     long getSequence() {
  58         return sequence;
  59     }
  60
  61     boolean acceptsResponse(final ResponseEnvelope<?> response) {
  62         return getSequence() == response.getSequence() && request.getTarget().equals(response.getMessage().getTarget());
  63     }
  64
  65     long getCurrentTry() {
  66         return lastTry.isPresent() ? lastTry.get().retry : 0;
  67      }
  68
  69     ClientActorBehavior complete(final Response<?, ?> response) {
  70         LOG.debug("Completing request {} with {}", request, response);
  71         return callback.complete(response);
  72     }
  73
  74     void poison(final RequestException cause) {
  75         LOG.trace("Poisoning request {}", request, cause);
  76         callback.complete(request.toRequestFailure(cause));
  77     }
  78
  79     boolean isTimedOut(final long now, final long timeoutNanos) {
  80         final long elapsed;
  81
  82         if (lastTry.isPresent()) {
  83             elapsed = now - lastTry.get().timeTicks;
  84         } else {
  85             elapsed = now - enqueuedTicks;
  86         }
  87
  88         if (elapsed >= timeoutNanos) {
  89             LOG.debug("Request {} timed out after {}ns", request, elapsed);
  90             return true;
  91         } else {
  92             return false;
  93         }
  94     }
  95
  96     void retransmit(final BackendInfo backend, final long now) {
  97         final long retry = lastTry.isPresent() ? lastTry.get().retry + 1 : 0;
  98         final RequestEnvelope toSend = new RequestEnvelope(request.toVersion(backend.getVersion()), sequence, retry);
  99
 100         final ActorRef actor = backend.getActor();
 101         LOG.trace("Retransmitting request {} as {} to {}", request, toSend, actor);
 102         actor.tell(toSend, ActorRef.noSender());
 103         lastTry = Optional.of(new LastTry(retry, now));
 104     }
 105
 106     @Override
 107     public String toString() {
 108         return MoreObjects.toStringHelper(SequencedQueueEntry.class).add("request", request).toString();
 109     }
 110 }