opendaylight/md-sal/cds-access-client/src/main/java/org/opendaylight/controller/cluster/access/client/AbstractReceivingClientConnection.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.Preconditions;
  12 import edu.umd.cs.findbugs.annotations.SuppressFBWarnings;
  13 import java.util.ArrayDeque;
  14 import java.util.Iterator;
  15 import java.util.Optional;
  16 import java.util.Queue;
  17 import javax.annotation.concurrent.GuardedBy;
  18 import org.opendaylight.controller.cluster.access.ABIVersion;
  19 import org.opendaylight.controller.cluster.access.concepts.Request;
  20 import org.opendaylight.controller.cluster.access.concepts.RequestException;
  21 import org.opendaylight.controller.cluster.access.concepts.Response;
  22 import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
  23 import org.slf4j.Logger;
  24 import org.slf4j.LoggerFactory;
  25 import scala.concurrent.duration.FiniteDuration;
  26
  27 /**
  28  * Implementation-internal intermediate subclass between {@link AbstractClientConnection} and two-out of three of its
  29  * sublcasses. It allows us to share some code.
  30  *
  31  * @author Robert Varga
  32  *
  33  * @param <T> Concrete {@link BackendInfo} type
  34  */
  35 abstract class AbstractReceivingClientConnection<T extends BackendInfo> extends AbstractClientConnection<T> {
  36     private static final Logger LOG = LoggerFactory.getLogger(AbstractReceivingClientConnection.class);
  37
  38     private final Queue<TransmittedConnectionEntry> inflight = new ArrayDeque<>();
  39     private final T backend;
  40
  41     private long lastProgress;
  42
  43     AbstractReceivingClientConnection(final ClientActorContext context, final Long cookie, final T backend) {
  44         super(context, cookie);
  45         this.backend = Preconditions.checkNotNull(backend);
  46         this.lastProgress = readTime();
  47     }
  48
  49     AbstractReceivingClientConnection(final AbstractReceivingClientConnection<T> oldConnection) {
  50         super(oldConnection);
  51         this.backend = oldConnection.backend;
  52         this.lastProgress = oldConnection.lastProgress;
  53     }
  54
  55     @Override
  56     public final Optional<T> getBackendInfo() {
  57         return Optional.of(backend);
  58     }
  59
  60     final ActorRef remoteActor() {
  61         return backend.getActor();
  62     }
  63
  64     final int remoteMaxMessages() {
  65         return backend.getMaxMessages();
  66     }
  67
  68     final ABIVersion remoteVersion() {
  69         return backend.getVersion();
  70     }
  71
  72     final long sessionId() {
  73         return backend.getSessionId();
  74     }
  75
  76     final int inflightSize() {
  77         return inflight.size();
  78     }
  79
  80     final void appendToInflight(final TransmittedConnectionEntry entry) {
  81         // This should never fail
  82         inflight.add(entry);
  83     }
  84
  85     @GuardedBy("this")
  86     @Override
  87     void spliceToSuccessor(final ReconnectForwarder successor) {
  88         ConnectionEntry entry = inflight.poll();
  89         while (entry != null) {
  90             successor.forwardEntry(entry);
  91             entry = inflight.poll();
  92         }
  93
  94         super.spliceToSuccessor(successor);
  95     }
  96
  97     @Override
  98     void receiveResponse(final ResponseEnvelope<?> envelope) {
  99         Optional<TransmittedConnectionEntry> maybeEntry = findMatchingEntry(inflight, envelope);
 100         if (maybeEntry == null) {
 101             LOG.debug("Request for {} not found in inflight queue, checking pending queue", envelope);
 102             maybeEntry = findMatchingEntry(pending(), envelope);
 103         }
 104
 105         if (maybeEntry == null || !maybeEntry.isPresent()) {
 106             LOG.warn("No request matching {} found, ignoring response", envelope);
 107             return;
 108         }
 109
 110         lastProgress = readTime();
 111         maybeEntry.get().complete(envelope.getMessage());
 112
 113         // We have freed up a slot, try to transmit something
 114         final int toSend = remoteMaxMessages() - inflight.size();
 115         if (toSend > 0) {
 116             sendMessages(toSend);
 117         }
 118     }
 119
 120     @Override
 121     boolean isEmpty() {
 122         return inflight.isEmpty() && super.isEmpty();
 123     }
 124
 125     @Override
 126     void poison(final RequestException cause) {
 127         super.poison(cause);
 128         poisonQueue(inflight, cause);
 129     }
 130
 131     /**
 132      * Transmit a given number of messages.
 133      *
 134      * @param count Number of messages to transmit, guaranteed to be positive.
 135      */
 136     abstract void sendMessages(int count);
 137
 138     /*
 139      * We are using tri-state return here to indicate one of three conditions:
 140      * - if a matching entry is found, return an Optional containing it
 141      * - if a matching entry is not found, but it makes sense to keep looking at other queues, return null
 142      * - if a conflicting entry is encountered, indicating we should ignore this request, return an empty Optional
 143      */
 144     @SuppressFBWarnings(value = "NP_OPTIONAL_RETURN_NULL",
 145             justification = "Returning null Optional is documented in the API contract.")
 146     private static Optional<TransmittedConnectionEntry> findMatchingEntry(final Queue<? extends ConnectionEntry> queue,
 147             final ResponseEnvelope<?> envelope) {
 148         // Try to find the request in a queue. Responses may legally come back in a different order, hence we need
 149         // to use an iterator
 150         final Iterator<? extends ConnectionEntry> it = queue.iterator();
 151         while (it.hasNext()) {
 152             final ConnectionEntry e = it.next();
 153             final Request<?, ?> request = e.getRequest();
 154             final Response<?, ?> response = envelope.getMessage();
 155
 156             // First check for matching target, or move to next entry
 157             if (!request.getTarget().equals(response.getTarget())) {
 158                 continue;
 159             }
 160
 161             // Sanity-check logical sequence, ignore any out-of-order messages
 162             if (request.getSequence() != response.getSequence()) {
 163                 LOG.debug("Expecting sequence {}, ignoring response {}", request.getSequence(), envelope);
 164                 return Optional.empty();
 165             }
 166
 167             // Check if the entry has (ever) been transmitted
 168             if (!(e instanceof TransmittedConnectionEntry)) {
 169                 return Optional.empty();
 170             }
 171
 172             final TransmittedConnectionEntry te = (TransmittedConnectionEntry) e;
 173
 174             // Now check session match
 175             if (envelope.getSessionId() != te.getSessionId()) {
 176                 LOG.debug("Expecting session {}, ignoring response {}", te.getSessionId(), envelope);
 177                 return Optional.empty();
 178             }
 179             if (envelope.getTxSequence() != te.getTxSequence()) {
 180                 LOG.warn("Expecting txSequence {}, ignoring response {}", te.getTxSequence(), envelope);
 181                 return Optional.empty();
 182             }
 183
 184             LOG.debug("Completing request {} with {}", request, envelope);
 185             it.remove();
 186             return Optional.of(te);
 187         }
 188
 189         return null;
 190     }
 191
 192     @SuppressFBWarnings(value = "NP_OPTIONAL_RETURN_NULL",
 193             justification = "Returning null Optional is documented in the API contract.")
 194     @Override
 195     final Optional<FiniteDuration> checkTimeout(final long now) {
 196         final Optional<FiniteDuration> xmit = checkTimeout(inflight.peek(), now);
 197         if (xmit == null) {
 198             return null;
 199         }
 200         final Optional<FiniteDuration> pend = super.checkTimeout(now);
 201         if (pend == null) {
 202             return null;
 203         }
 204         if (!xmit.isPresent()) {
 205             return pend;
 206         }
 207         if (!pend.isPresent()) {
 208             return xmit;
 209         }
 210
 211         return Optional.of(xmit.get().min(pend.get()));
 212     }
 213 }