BUG-5280: introduce request/response Envelope

[controller.git] / opendaylight / md-sal / sal-distributed-datastore / src / main / java / org / opendaylight / controller / cluster / datastore / actors / client / SequencedQueue.java

diff --git a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/actors/client/SequencedQueue.java b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/actors/client/SequencedQueue.java
index 8cae0e133a58edd9dba686c651850c950c7aafad..50c600f52d2f9be7dbfcfee1e6147f3dd03db0b2 100644 (file)
--- a/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/actors/client/SequencedQueue.java
+++ b/opendaylight/md-sal/sal-distributed-datastore/src/main/java/org/opendaylight/controller/cluster/datastore/actors/client/SequencedQueue.java
@@ -20,7 +20,7 @@ import javax.annotation.Nullable;
 import javax.annotation.concurrent.NotThreadSafe;
 import org.opendaylight.controller.cluster.access.concepts.Request;
 import org.opendaylight.controller.cluster.access.concepts.RequestException;
-import org.opendaylight.controller.cluster.access.concepts.Response;
+import org.opendaylight.controller.cluster.access.concepts.ResponseEnvelope;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import scala.concurrent.duration.FiniteDuration;
@@ -105,12 +105,13 @@ final class SequencedQueue {
      * @param callback Callback to be invoked
      * @return Optional duration with semantics described above.
      */
-    @Nullable Optional<FiniteDuration> enqueueRequest(final Request<?, ?> request, final RequestCallback callback) {
+    @Nullable Optional<FiniteDuration> enqueueRequest(final long sequence, final Request<?, ?> request,
+            final RequestCallback callback) {
+        checkNotClosed();
+
         final long now = ticker.read();
-        final SequencedQueueEntry e = new SequencedQueueEntry(request, callback, now);
+        final SequencedQueueEntry e = new SequencedQueueEntry(request, sequence, callback, now);
 
-        // We could have check first, but argument checking needs to happen first
-        checkNotClosed();
         queue.add(e);
         LOG.debug("Enqueued request {} to queue {}", request, this);
 
@@ -127,7 +128,7 @@ final class SequencedQueue {
         }
     }
 
-    ClientActorBehavior complete(final ClientActorBehavior current, final Response<?, ?> response) {
+    ClientActorBehavior complete(final ClientActorBehavior current, final ResponseEnvelope<?> response) {
         // Responses to different targets may arrive out of order, hence we use an iterator
         final Iterator<SequencedQueueEntry> it = queue.iterator();
         while (it.hasNext()) {
@@ -136,7 +137,7 @@ final class SequencedQueue {
                 lastProgress = ticker.read();
                 it.remove();
                 LOG.debug("Completing request {} with {}", e, response);
-                return e.complete(response);
+                return e.complete(response.getMessage());
             }
         }