Added support for responding directly via a cluster message received from a peer

 package org.onlab.onos.store.cluster.messaging;
 package org.onlab.onos.store.cluster.messaging;
 
 
+import java.io.IOException;
+
 import org.onlab.onos.cluster.NodeId;
 import org.onlab.onos.cluster.NodeId;
 
 
 // TODO: Should payload type be ByteBuffer?
 // TODO: Should payload type be ByteBuffer?
@@ -49,4 +51,14 @@ public class ClusterMessage {
     public byte[] payload() {
     public byte[] payload() {
         return payload;
         return payload;
     }
     }
+
+    /**
+     * Sends a response to the sender.
+     *
+     * @param data payload response.
+     * @throws IOException
+     */
+    public void respond(byte[] data) throws IOException {
+        throw new IllegalStateException("One can only repond to message recived from others.");
+    }
 }
 }

@@ -158,7 +158,7 @@ public class ClusterCommunicationManager
         public void handle(Message message) {
         public void handle(Message message) {
             try {
             try {
                 ClusterMessage clusterMessage = SERIALIZER.decode(message.payload());
                 ClusterMessage clusterMessage = SERIALIZER.decode(message.payload());
-                handler.handle(clusterMessage);
+                handler.handle(new InternalClusterMessage(clusterMessage, message));
             } catch (Exception e) {
             } catch (Exception e) {
                 log.error("Exception caught during ClusterMessageHandler", e);
                 log.error("Exception caught during ClusterMessageHandler", e);
                 throw e;
                 throw e;
@@ -166,6 +166,21 @@ public class ClusterCommunicationManager
         }
         }
     }
     }
 
 
+    public static final class InternalClusterMessage extends ClusterMessage {
+
+        private final Message rawMessage;
+
+        public InternalClusterMessage(ClusterMessage clusterMessage, Message rawMessage) {
+            super(clusterMessage.sender(), clusterMessage.subject(), clusterMessage.payload());
+            this.rawMessage = rawMessage;
+        }
+
+        @Override
+        public void respond(byte[] response) throws IOException {
+            rawMessage.respond(response);
+        }
+    }
+
     private static final class InternalClusterMessageResponse implements ClusterMessageResponse {
     private static final class InternalClusterMessageResponse implements ClusterMessageResponse {
 
 
         private final NodeId sender;
         private final NodeId sender;