Added sendAndReceive API to ClusterCommunicationService

@@ -37,6 +37,15 @@ public interface ClusterCommunicationService {
     boolean multicast(ClusterMessage message, Set<NodeId> nodeIds) throws IOException;
     boolean multicast(ClusterMessage message, Set<NodeId> nodeIds) throws IOException;
 
 
     /**
     /**
+     * Sends a message synchronously.
+     * @param message message to send
+     * @param toNodeId recipient node identifier
+     * @return ClusterMessageResponse which is reply future.
+     * @throws IOException
+     */
+    ClusterMessageResponse sendAndReceive(ClusterMessage message, NodeId toNodeId) throws IOException;
+
+    /**
      * Adds a new subscriber for the specified message subject.
      * Adds a new subscriber for the specified message subject.
      *
      *
      * @param subject    message subject
      * @param subject    message subject

+package org.onlab.onos.store.cluster.messaging;
+
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
+import org.onlab.onos.cluster.NodeId;
+
+public interface ClusterMessageResponse {
+    public NodeId sender();
+    public byte[] get(long timeout, TimeUnit timeunit) throws TimeoutException;
+    public byte[] get(long timeout) throws InterruptedException;
+}

@@ -4,6 +4,9 @@ import static com.google.common.base.Preconditions.checkArgument;
 
 
 import java.io.IOException;
 import java.io.IOException;
 import java.util.Set;
 import java.util.Set;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
+
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Component;
 import org.apache.felix.scr.annotations.Deactivate;
 import org.apache.felix.scr.annotations.Deactivate;
@@ -17,6 +20,7 @@ import org.onlab.onos.store.cluster.impl.ClusterMembershipEvent;
 import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
 import org.onlab.onos.store.cluster.messaging.ClusterCommunicationService;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessage;
 import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
 import org.onlab.onos.store.cluster.messaging.ClusterMessageHandler;
+import org.onlab.onos.store.cluster.messaging.ClusterMessageResponse;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.cluster.messaging.MessageSubject;
 import org.onlab.onos.store.serializers.ClusterMessageSerializer;
 import org.onlab.onos.store.serializers.ClusterMessageSerializer;
 import org.onlab.onos.store.serializers.KryoPoolUtil;
 import org.onlab.onos.store.serializers.KryoPoolUtil;
@@ -28,6 +32,7 @@ import org.onlab.netty.Message;
 import org.onlab.netty.MessageHandler;
 import org.onlab.netty.MessageHandler;
 import org.onlab.netty.MessagingService;
 import org.onlab.netty.MessagingService;
 import org.onlab.netty.NettyMessagingService;
 import org.onlab.netty.NettyMessagingService;
+import org.onlab.netty.Response;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
@@ -120,6 +125,22 @@ public class ClusterCommunicationManager
     }
     }
 
 
     @Override
     @Override
+    public ClusterMessageResponse sendAndReceive(ClusterMessage message, NodeId toNodeId) throws IOException {
+        ControllerNode node = clusterService.getNode(toNodeId);
+        checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
+        Endpoint nodeEp = new Endpoint(node.ip().toString(), node.tcpPort());
+        try {
+            Response responseFuture =
+                    messagingService.sendAndReceive(nodeEp, message.subject().value(), SERIALIZER.encode(message));
+            return new InternalClusterMessageResponse(toNodeId, responseFuture);
+
+        } catch (IOException e) {
+            log.error("Failed interaction with remote nodeId: " + toNodeId, e);
+            throw e;
+        }
+    }
+
+    @Override
     public void addSubscriber(MessageSubject subject,
     public void addSubscriber(MessageSubject subject,
             ClusterMessageHandler subscriber) {
             ClusterMessageHandler subscriber) {
         messagingService.registerHandler(subject.value(), new InternalClusterMessageHandler(subscriber));
         messagingService.registerHandler(subject.value(), new InternalClusterMessageHandler(subscriber));
@@ -144,4 +165,30 @@ public class ClusterCommunicationManager
             }
             }
         }
         }
     }
     }
+
+    private static final class InternalClusterMessageResponse implements ClusterMessageResponse {
+
+        private final NodeId sender;
+        private final Response responseFuture;
+
+        public InternalClusterMessageResponse(NodeId sender, Response responseFuture) {
+            this.sender = sender;
+            this.responseFuture = responseFuture;
+        }
+        @Override
+        public NodeId sender() {
+            return sender;
+        }
+
+        @Override
+        public byte[] get(long timeout, TimeUnit timeunit)
+                throws TimeoutException {
+            return responseFuture.get(timeout, timeunit);
+        }
+
+        @Override
+        public byte[] get(long timeout) throws InterruptedException {
+            return responseFuture.get();
+        }
+    }
 }
 }