Revamped ClusterCommunicationService API

Change-Id: I9326369de3d2413b0882b324979d10483c093de9

@@ -219,7 +219,7 @@ public class IntentPerfCollector {
 
 
     private void broadcastSample(long time, NodeId nodeId, double overallRate, double currentRate) {
     private void broadcastSample(long time, NodeId nodeId, double overallRate, double currentRate) {
         String data = String.format("%d|%f|%f", time, overallRate, currentRate);
         String data = String.format("%d|%f|%f", time, overallRate, currentRate);
-        communicationService.broadcast(new ClusterMessage(nodeId, SAMPLE, data.getBytes()));
+        communicationService.broadcast(data, SAMPLE, str -> str.getBytes());
     }
     }
 
 
     private class InternalSampleCollector implements ClusterMessageHandler {
     private class InternalSampleCollector implements ClusterMessageHandler {

@@ -249,14 +249,14 @@ public class IntentPerfInstaller {
     public void start() {
     public void start() {
         if (stopped) {
         if (stopped) {
             stopped = false;
             stopped = false;
-            communicationService.broadcast(new ClusterMessage(nodeId, CONTROL, START.getBytes()));
+            communicationService.broadcast(START, CONTROL, str -> str.getBytes());
             startTestRun();
             startTestRun();
         }
         }
     }
     }
 
 
     public void stop() {
     public void stop() {
         if (!stopped) {
         if (!stopped) {
-            communicationService.broadcast(new ClusterMessage(nodeId, CONTROL, STOP.getBytes()));
+            communicationService.broadcast(STOP, CONTROL, str -> str.getBytes());
             stopTestRun();
             stopTestRun();
         }
         }
     }
     }

@@ -15,13 +15,16 @@
  */
  */
 package org.onosproject.store.cluster.messaging;
 package org.onosproject.store.cluster.messaging;
 
 
-import com.google.common.util.concurrent.ListenableFuture;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+import java.util.function.Function;
+
 import org.onosproject.cluster.NodeId;
 import org.onosproject.cluster.NodeId;
 
 
-import java.io.IOException;
-import java.util.concurrent.ExecutorService;
+import com.google.common.util.concurrent.ListenableFuture;
 
 
-// TODO: remove IOExceptions?
 /**
 /**
  * Service for assisting communications between controller cluster nodes.
  * Service for assisting communications between controller cluster nodes.
  */
  */
@@ -33,6 +36,7 @@ public interface ClusterCommunicationService {
      * @param message  message to send
      * @param message  message to send
      * @return true if the message was sent successfully to all nodes; false otherwise.
      * @return true if the message was sent successfully to all nodes; false otherwise.
      */
      */
+    @Deprecated
     boolean broadcast(ClusterMessage message);
     boolean broadcast(ClusterMessage message);
 
 
     /**
     /**
@@ -41,6 +45,7 @@ public interface ClusterCommunicationService {
      * @param message  message to send
      * @param message  message to send
      * @return true if the message was sent successfully to all nodes; false otherwise.
      * @return true if the message was sent successfully to all nodes; false otherwise.
      */
      */
+    @Deprecated
     boolean broadcastIncludeSelf(ClusterMessage message);
     boolean broadcastIncludeSelf(ClusterMessage message);
 
 
     /**
     /**
@@ -50,6 +55,7 @@ public interface ClusterCommunicationService {
      * @param toNodeId node identifier
      * @param toNodeId node identifier
      * @return true if the message was sent successfully; false otherwise.
      * @return true if the message was sent successfully; false otherwise.
      */
      */
+    @Deprecated
     boolean unicast(ClusterMessage message, NodeId toNodeId);
     boolean unicast(ClusterMessage message, NodeId toNodeId);
 
 
     /**
     /**
@@ -59,6 +65,7 @@ public interface ClusterCommunicationService {
      * @param nodeIds  recipient node identifiers
      * @param nodeIds  recipient node identifiers
      * @return true if the message was sent successfully to all nodes in the group; false otherwise.
      * @return true if the message was sent successfully to all nodes in the group; false otherwise.
      */
      */
+    @Deprecated
     boolean multicast(ClusterMessage message, Iterable<NodeId> nodeIds);
     boolean multicast(ClusterMessage message, Iterable<NodeId> nodeIds);
 
 
     /**
     /**
@@ -66,27 +73,121 @@ public interface ClusterCommunicationService {
      * @param message message to send
      * @param message message to send
      * @param toNodeId recipient node identifier
      * @param toNodeId recipient node identifier
      * @return reply future.
      * @return reply future.
-     * @throws IOException when I/O exception of some sort has occurred
      */
      */
-    ListenableFuture<byte[]> sendAndReceive(ClusterMessage message, NodeId toNodeId) throws IOException;
+    @Deprecated
+    ListenableFuture<byte[]> sendAndReceive(ClusterMessage message, NodeId toNodeId);
 
 
     /**
     /**
      * 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
      * @param subscriber message subscriber
      * @param subscriber message subscriber
+     * @param executor executor to use for running handler.
      */
      */
     @Deprecated
     @Deprecated
-    void addSubscriber(MessageSubject subject, ClusterMessageHandler subscriber);
+    void addSubscriber(MessageSubject subject, ClusterMessageHandler subscriber, ExecutorService executor);
+
+    /**
+     * Broadcasts a message to all controller nodes.
+     *
+     * @param message message to send
+     * @param subject message subject
+     * @param encoder function for encoding message to byte[]
+     * @param <M> message type
+     */
+    <M> void broadcast(M message,
+                       MessageSubject subject,
+                       Function<M, byte[]> encoder);
+
+    /**
+     * Broadcasts a message to all controller nodes including self.
+     *
+     * @param message message to send
+     * @param subject message subject
+     * @param encoder function for encoding message to byte[]
+     * @param <M> message type
+     */
+    <M> void broadcastIncludeSelf(M message,
+                                  MessageSubject subject,
+                                  Function<M, byte[]> encoder);
+
+    /**
+     * Sends a message to the specified controller node.
+     *
+     * @param message message to send
+     * @param subject message subject
+     * @param encoder function for encoding message to byte[]
+     * @param toNodeId destination node identifier
+     * @param <M> message type
+     * @return true if the message was sent successfully; false otherwise
+     */
+    <M> boolean unicast(M message,
+                        MessageSubject subject,
+                        Function<M, byte[]> encoder,
+                        NodeId toNodeId);
+
+    /**
+     * Multicasts a message to a set of controller nodes.
+     *
+     * @param message message to send
+     * @param subject message subject
+     * @param encoder function for encoding message to byte[]
+     * @param nodeIds  recipient node identifiers
+     * @param <M> message type
+     */
+    <M> void multicast(M message,
+                       MessageSubject subject,
+                       Function<M, byte[]> encoder,
+                       Set<NodeId> nodeIds);
+
+    /**
+     * Sends a message and expects a reply.
+     *
+     * @param message message to send
+     * @param subject message subject
+     * @param encoder function for encoding request to byte[]
+     * @param decoder function for decoding response from byte[]
+     * @param toNodeId recipient node identifier
+     * @param <M> request type
+     * @param <R> reply type
+     * @return reply future
+     */
+    <M, R> CompletableFuture<R> sendAndReceive(M message,
+                                               MessageSubject subject,
+                                               Function<M, byte[]> encoder,
+                                               Function<byte[], R> decoder,
+                                               NodeId toNodeId);
 
 
     /**
     /**
      * 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
-     * @param subscriber message subscriber
-     * @param executor executor to use for running handler.
+     * @param decoder decoder for resurrecting incoming message
+     * @param handler handler function that process the incoming message and produces a reply
+     * @param encoder encoder for serializing reply
+     * @param executor executor to run this handler on
+     * @param <M> incoming message type
+     * @param <R> reply message type
      */
      */
-    void addSubscriber(MessageSubject subject, ClusterMessageHandler subscriber, ExecutorService executor);
+    <M, R> void addSubscriber(MessageSubject subject,
+                              Function<byte[], M> decoder,
+                              Function<M, R> handler,
+                              Function<R, byte[]> encoder,
+                              ExecutorService executor);
+
+    /**
+     * Adds a new subscriber for the specified message subject.
+     *
+     * @param subject message subject
+     * @param decoder decoder to resurrecting incoming message
+     * @param handler handler for handling message
+     * @param executor executor to run this handler on
+     * @param <M> incoming message type
+     */
+    <M> void addSubscriber(MessageSubject subject,
+                           Function<byte[], M> decoder,
+                           Consumer<M> handler,
+                           ExecutorService executor);
 
 
     /**
     /**
      * Removes a subscriber for the specified message subject.
      * Removes a subscriber for the specified message subject.
@@ -94,5 +195,4 @@ public interface ClusterCommunicationService {
      * @param subject message subject
      * @param subject message subject
      */
      */
     void removeSubscriber(MessageSubject subject);
     void removeSubscriber(MessageSubject subject);
-
 }
 }

@@ -17,7 +17,6 @@ package org.onosproject.store.app;
 
 
 import com.google.common.base.Charsets;
 import com.google.common.base.Charsets;
 import com.google.common.collect.ImmutableSet;
 import com.google.common.collect.ImmutableSet;
-import com.google.common.util.concurrent.ListenableFuture;
 
 
 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;
@@ -54,13 +53,14 @@ import org.onosproject.store.service.StorageService;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 
 
 import java.io.ByteArrayInputStream;
 import java.io.ByteArrayInputStream;
-import java.io.IOException;
 import java.io.InputStream;
 import java.io.InputStream;
 import java.util.Set;
 import java.util.Set;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Executors;
 import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.ScheduledExecutorService;
+import java.util.function.Function;
+
 import static com.google.common.io.ByteStreams.toByteArray;
 import static com.google.common.io.ByteStreams.toByteArray;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static java.util.concurrent.TimeUnit.MILLISECONDS;
 import static org.onlab.util.Tools.groupedThreads;
 import static org.onlab.util.Tools.groupedThreads;
@@ -351,22 +351,34 @@ public class GossipApplicationStore extends ApplicationArchive
      */
      */
     private void fetchBits(Application app) {
     private void fetchBits(Application app) {
         ControllerNode localNode = clusterService.getLocalNode();
         ControllerNode localNode = clusterService.getLocalNode();
-        ClusterMessage message = new ClusterMessage(localNode.id(), APP_BITS_REQUEST,
-                                                    app.id().name().getBytes(Charsets.UTF_8));
-        //Map<ControllerNode, ListenableFuture<byte[]>> futures = new HashMap<>();
         CountDownLatch latch = new CountDownLatch(1);
         CountDownLatch latch = new CountDownLatch(1);
 
 
         // FIXME: send message with name & version to make sure we don't get served old bits
         // FIXME: send message with name & version to make sure we don't get served old bits
 
 
         log.info("Downloading bits for application {}", app.id().name());
         log.info("Downloading bits for application {}", app.id().name());
         for (ControllerNode node : clusterService.getNodes()) {
         for (ControllerNode node : clusterService.getNodes()) {
-            try {
-                ListenableFuture<byte[]> future = clusterCommunicator.sendAndReceive(message, node.id());
-                future.addListener(new InternalBitListener(app, node, future, latch), executor);
-            } catch (IOException e) {
-                log.debug("Unable to request bits for application {} from node {}",
+            if (latch.getCount() == 0) {
+                break;
+            }
+            if (node.equals(localNode)) {
+                continue;
+            }
+            clusterCommunicator.sendAndReceive(app.id().name(),
+                                    APP_BITS_REQUEST,
+                                    s -> s.getBytes(Charsets.UTF_8),
+                                    Function.identity(),
+                                    node.id())
+                               .whenCompleteAsync((bits, error) -> {
+                                   if (error == null && latch.getCount() > 0) {
+                                       saveApplication(new ByteArrayInputStream(bits));
+                                       log.info("Downloaded bits for application {} from node {}",
                                                app.id().name(), node.id());
                                                app.id().name(), node.id());
+                                       latch.countDown();
+                                   } else if (error != null) {
+                                       log.warn("Unable to fetch bits for application {} from node {}",
+                                               app.id().name(), node.id(), error);
                                    }
                                    }
+                               }, executor);
         }
         }
 
 
         try {
         try {
@@ -392,41 +404,6 @@ public class GossipApplicationStore extends ApplicationArchive
             }
             }
         }
         }
     }
     }
-
-    /**
-     * Processes completed fetch requests.
-     */
-    private class InternalBitListener implements Runnable {
-        private final Application app;
-        private final ControllerNode node;
-        private final ListenableFuture<byte[]> future;
-        private final CountDownLatch latch;
-
-        public InternalBitListener(Application app, ControllerNode node,
-                                   ListenableFuture<byte[]> future, CountDownLatch latch) {
-            this.app = app;
-            this.node = node;
-            this.future = future;
-            this.latch = latch;
-        }
-
-        @Override
-        public void run() {
-            if (latch.getCount() > 0 && !future.isCancelled()) {
-                try {
-                    byte[] bits = future.get(1, MILLISECONDS);
-                    saveApplication(new ByteArrayInputStream(bits));
-                    log.info("Downloaded bits for application {} from node {}",
-                             app.id().name(), node.id());
-                    latch.countDown();
-                } catch (Exception e) {
-                    log.warn("Unable to fetch bits for application {} from node {}",
-                             app.id().name(), node.id());
-                }
-            }
-        }
-    }
-
     /**
     /**
      * Prunes applications which are not in the map, but are on disk.
      * Prunes applications which are not in the map, but are on disk.
      */
      */
@@ -449,6 +426,4 @@ public class GossipApplicationStore extends ApplicationArchive
                                       appDesc.origin(), appDesc.permissions(),
                                       appDesc.origin(), appDesc.permissions(),
                                       appDesc.featuresRepo(), appDesc.features());
                                       appDesc.featuresRepo(), appDesc.features());
     }
     }
-
 }
 }
-

@@ -419,10 +419,9 @@ public class HazelcastLeadershipService implements LeadershipService {
                             // Dispatch to all instances
                             // Dispatch to all instances
 
 
                             clusterCommunicator.broadcastIncludeSelf(
                             clusterCommunicator.broadcastIncludeSelf(
-                                    new ClusterMessage(
-                                            clusterService.getLocalNode().id(),
+                                    leadershipEvent,
                                     LEADERSHIP_EVENT_MESSAGE_SUBJECT,
                                     LEADERSHIP_EVENT_MESSAGE_SUBJECT,
-                                            SERIALIZER.encode(leadershipEvent)));
+                                    SERIALIZER::encode);
                         } else {
                         } else {
                             //
                             //
                             // Test if time to expire a stale leader
                             // Test if time to expire a stale leader
@@ -491,11 +490,11 @@ public class HazelcastLeadershipService implements LeadershipService {
                         leadershipEvent = new LeadershipEvent(
                         leadershipEvent = new LeadershipEvent(
                                              LeadershipEvent.Type.LEADER_ELECTED,
                                              LeadershipEvent.Type.LEADER_ELECTED,
                                              new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
                                              new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
+
                         clusterCommunicator.broadcastIncludeSelf(
                         clusterCommunicator.broadcastIncludeSelf(
-                                new ClusterMessage(
-                                        clusterService.getLocalNode().id(),
+                                leadershipEvent,
                                 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
                                 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
-                                        SERIALIZER.encode(leadershipEvent)));
+                                SERIALIZER::encode);
                     }
                     }
 
 
                     // Sleep forever until interrupted
                     // Sleep forever until interrupted
@@ -519,11 +518,12 @@ public class HazelcastLeadershipService implements LeadershipService {
                         leadershipEvent = new LeadershipEvent(
                         leadershipEvent = new LeadershipEvent(
                                                  LeadershipEvent.Type.LEADER_BOOTED,
                                                  LeadershipEvent.Type.LEADER_BOOTED,
                                                  new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
                                                  new Leadership(topicName, localNodeId, myLastLeaderTerm, 0));
+
                         clusterCommunicator.broadcastIncludeSelf(
                         clusterCommunicator.broadcastIncludeSelf(
-                                new ClusterMessage(
-                                        clusterService.getLocalNode().id(),
+                                leadershipEvent,
                                 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
                                 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
-                                        SERIALIZER.encode(leadershipEvent)));
+                                SERIALIZER::encode);
+
                         if (leaderLock.isLockedByCurrentThread()) {
                         if (leaderLock.isLockedByCurrentThread()) {
                             leaderLock.unlock();
                             leaderLock.unlock();
                         }
                         }

@@ -15,7 +15,6 @@
  */
  */
 package org.onosproject.store.cluster.messaging.impl;
 package org.onosproject.store.cluster.messaging.impl;
 
 
-import com.google.common.util.concurrent.ListenableFuture;
 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;
@@ -37,8 +36,17 @@ import org.onosproject.store.cluster.messaging.MessageSubject;
 import org.slf4j.Logger;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 import org.slf4j.LoggerFactory;
 
 
+import com.google.common.base.Objects;
+import com.google.common.util.concurrent.ListenableFuture;
+import com.google.common.util.concurrent.SettableFuture;
+
 import java.io.IOException;
 import java.io.IOException;
+import java.util.Set;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
+import java.util.function.Consumer;
+import java.util.function.Function;
+import java.util.stream.Collectors;
 
 
 import static com.google.common.base.Preconditions.checkArgument;
 import static com.google.common.base.Preconditions.checkArgument;
 
 
@@ -122,46 +130,101 @@ public class ClusterCommunicationManager
         return unicastUnchecked(message.subject(), message.getBytes(), toNodeId);
         return unicastUnchecked(message.subject(), message.getBytes(), toNodeId);
     }
     }
 
 
-    private boolean unicast(MessageSubject subject, byte[] payload, NodeId toNodeId) throws IOException {
-        ControllerNode node = clusterService.getNode(toNodeId);
-        checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
-        Endpoint nodeEp = new Endpoint(node.ip(), node.tcpPort());
-        try {
-            messagingService.sendAsync(nodeEp, subject.value(), payload);
-            return true;
-        } catch (IOException e) {
-            log.debug("Failed to send cluster message to nodeId: " + toNodeId, e);
-            throw e;
+    @Override
+    public ListenableFuture<byte[]> sendAndReceive(ClusterMessage message, NodeId toNodeId) {
+        SettableFuture<byte[]> response = SettableFuture.create();
+        sendAndReceive(message.subject(), message.getBytes(), toNodeId).whenComplete((r, e) -> {
+            if (e == null) {
+                response.set(r);
+            } else {
+                response.setException(e);
             }
             }
+        });
+        return response;
     }
     }
 
 
-    private boolean unicastUnchecked(MessageSubject subject, byte[] payload, NodeId toNodeId) {
-        try {
-            return unicast(subject, payload, toNodeId);
-        } catch (IOException e) {
-            return false;
+    @Override
+    public <M> void broadcast(M message,
+                              MessageSubject subject,
+                              Function<M, byte[]> encoder) {
+        multicast(message,
+                  subject,
+                  encoder,
+                  clusterService.getNodes()
+                      .stream()
+                      .filter(node -> !Objects.equal(node, clusterService.getLocalNode()))
+                      .map(ControllerNode::id)
+                      .collect(Collectors.toSet()));
+    }
+
+    @Override
+    public <M> void broadcastIncludeSelf(M message,
+                                         MessageSubject subject,
+                                         Function<M, byte[]> encoder) {
+        multicast(message,
+                  subject,
+                  encoder,
+                  clusterService.getNodes()
+                      .stream()
+                      .map(ControllerNode::id)
+                      .collect(Collectors.toSet()));
     }
     }
+
+    @Override
+    public <M> boolean unicast(M message,
+                               MessageSubject subject,
+                               Function<M, byte[]> encoder,
+                               NodeId toNodeId) {
+        byte[] payload = new ClusterMessage(
+                clusterService.getLocalNode().id(),
+                subject,
+                encoder.apply(message)).getBytes();
+        return unicastUnchecked(subject, payload, toNodeId);
     }
     }
 
 
     @Override
     @Override
-    public ListenableFuture<byte[]> sendAndReceive(ClusterMessage message, NodeId toNodeId) throws IOException {
+    public <M> void multicast(M message,
+                              MessageSubject subject,
+                              Function<M, byte[]> encoder,
+                              Set<NodeId> nodes) {
+        byte[] payload = new ClusterMessage(
+                clusterService.getLocalNode().id(),
+                subject,
+                encoder.apply(message)).getBytes();
+        nodes.forEach(nodeId -> unicastUnchecked(subject, payload, nodeId));
+    }
+
+    @Override
+    public <M, R> CompletableFuture<R> sendAndReceive(M message,
+                                                      MessageSubject subject,
+                                                      Function<M, byte[]> encoder,
+                                                      Function<byte[], R> decoder,
+                                                      NodeId toNodeId) {
+        ClusterMessage envelope = new ClusterMessage(
+                clusterService.getLocalNode().id(),
+                subject,
+                encoder.apply(message));
+        return sendAndReceive(subject, envelope.getBytes(), toNodeId).thenApply(decoder);
+    }
+
+    private boolean unicastUnchecked(MessageSubject subject, byte[] payload, NodeId toNodeId) {
         ControllerNode node = clusterService.getNode(toNodeId);
         ControllerNode node = clusterService.getNode(toNodeId);
         checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
         checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
         Endpoint nodeEp = new Endpoint(node.ip(), node.tcpPort());
         Endpoint nodeEp = new Endpoint(node.ip(), node.tcpPort());
         try {
         try {
-            return messagingService.sendAndReceive(nodeEp, message.subject().value(), message.getBytes());
-
+            messagingService.sendAsync(nodeEp, subject.value(), payload);
+            return true;
         } catch (IOException e) {
         } catch (IOException e) {
-            log.trace("Failed interaction with remote nodeId: " + toNodeId, e);
-            throw e;
+            log.debug("Failed to send cluster message to nodeId: " + toNodeId, e);
+            return false;
         }
         }
     }
     }
 
 
-    @Override
-    @Deprecated
-    public void addSubscriber(MessageSubject subject,
-                              ClusterMessageHandler subscriber) {
-        messagingService.registerHandler(subject.value(), new InternalClusterMessageHandler(subscriber));
+    private CompletableFuture<byte[]> sendAndReceive(MessageSubject subject, byte[] payload, NodeId toNodeId) {
+        ControllerNode node = clusterService.getNode(toNodeId);
+        checkArgument(node != null, "Unknown nodeId: %s", toNodeId);
+        Endpoint nodeEp = new Endpoint(node.ip(), node.tcpPort());
+        return messagingService.sendAndReceive(nodeEp, subject.value(), payload);
     }
     }
 
 
     @Override
     @Override
@@ -202,6 +265,60 @@ public class ClusterCommunicationManager
         }
         }
     }
     }
 
 
+    @Override
+    public <M, R> void addSubscriber(MessageSubject subject,
+            Function<byte[], M> decoder,
+            Function<M, R> handler,
+            Function<R, byte[]> encoder,
+            ExecutorService executor) {
+        messagingService.registerHandler(subject.value(),
+                new InternalMessageResponder<>(decoder, encoder, handler),
+                executor);
+    }
+
+    @Override
+    public <M> void addSubscriber(MessageSubject subject,
+            Function<byte[], M> decoder,
+            Consumer<M> handler,
+            ExecutorService executor) {
+        messagingService.registerHandler(subject.value(),
+                new InternalMessageConsumer<>(decoder, handler),
+                executor);
+    }
+
+    private class InternalMessageResponder<M, R> implements MessageHandler {
+        private final Function<byte[], M> decoder;
+        private final Function<R, byte[]> encoder;
+        private final Function<M, R> handler;
+
+        public InternalMessageResponder(Function<byte[], M> decoder,
+                                        Function<R, byte[]> encoder,
+                                        Function<M, R> handler) {
+            this.decoder = decoder;
+            this.encoder = encoder;
+            this.handler = handler;
+        }
+        @Override
+        public void handle(Message message) throws IOException {
+            R response = handler.apply(decoder.apply(ClusterMessage.fromBytes(message.payload()).payload()));
+            message.respond(encoder.apply(response));
+        }
+    }
+
+    private class InternalMessageConsumer<M> implements MessageHandler {
+        private final Function<byte[], M> decoder;
+        private final Consumer<M> consumer;
+
+        public InternalMessageConsumer(Function<byte[], M> decoder, Consumer<M> consumer) {
+            this.decoder = decoder;
+            this.consumer = consumer;
+        }
+        @Override
+        public void handle(Message message) throws IOException {
+            consumer.accept(decoder.apply(ClusterMessage.fromBytes(message.payload()).payload()));
+        }
+    }
+
     public static final class InternalClusterMessage extends ClusterMessage {
     public static final class InternalClusterMessage extends ClusterMessage {
 
 
         private final Message rawMessage;
         private final Message rawMessage;

@@ -343,11 +343,9 @@ public class DistributedLeadershipManager implements LeadershipService {
 
 
     private void notifyPeers(LeadershipEvent event) {
     private void notifyPeers(LeadershipEvent event) {
         eventDispatcher.post(event);
         eventDispatcher.post(event);
-        clusterCommunicator.broadcast(
-                new ClusterMessage(
-                        clusterService.getLocalNode().id(),
+        clusterCommunicator.broadcast(event,
                 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
                 LEADERSHIP_EVENT_MESSAGE_SUBJECT,
-                        SERIALIZER.encode(event)));
+                SERIALIZER::encode);
     }
     }
 
 
     private void notifyRemovedLeader(String path, NodeId leader, long epoch, long electedTime) {
     private void notifyRemovedLeader(String path, NodeId leader, long epoch, long electedTime) {
@@ -366,11 +364,9 @@ public class DistributedLeadershipManager implements LeadershipService {
         if (updatedLeader) {
         if (updatedLeader) {
             LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_BOOTED, oldLeadership);
             LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_BOOTED, oldLeadership);
             eventDispatcher.post(event);
             eventDispatcher.post(event);
-            clusterCommunicator.broadcast(
-                    new ClusterMessage(
-                            clusterService.getLocalNode().id(),
+            clusterCommunicator.broadcast(event,
                     LEADERSHIP_EVENT_MESSAGE_SUBJECT,
                     LEADERSHIP_EVENT_MESSAGE_SUBJECT,
-                            SERIALIZER.encode(event)));
+                    SERIALIZER::encode);
         }
         }
     }
     }
 
 
@@ -469,11 +465,9 @@ public class DistributedLeadershipManager implements LeadershipService {
             leaderBoard.forEach((path, leadership) -> {
             leaderBoard.forEach((path, leadership) -> {
                 if (leadership.leader().equals(localNodeId)) {
                 if (leadership.leader().equals(localNodeId)) {
                     LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, leadership);
                     LeadershipEvent event = new LeadershipEvent(LeadershipEvent.Type.LEADER_ELECTED, leadership);
-                    clusterCommunicator.broadcast(
-                            new ClusterMessage(
-                                    clusterService.getLocalNode().id(),
+                    clusterCommunicator.broadcast(event,
                             LEADERSHIP_EVENT_MESSAGE_SUBJECT,
                             LEADERSHIP_EVENT_MESSAGE_SUBJECT,
-                                    SERIALIZER.encode(event)));
+                            SERIALIZER::encode);
                 }
                 }
             });
             });
         } catch (Exception e) {
         } catch (Exception e) {

@@ -304,11 +304,9 @@ public class GossipDeviceStore
 
 
             DeviceInjectedEvent deviceInjectedEvent = new DeviceInjectedEvent(
             DeviceInjectedEvent deviceInjectedEvent = new DeviceInjectedEvent(
                     providerId, deviceId, deviceDescription);
                     providerId, deviceId, deviceDescription);
-            ClusterMessage clusterMessage = new ClusterMessage(localNode, DEVICE_INJECTED,
-                    SERIALIZER.encode(deviceInjectedEvent));
 
 
             // TODO check unicast return value
             // TODO check unicast return value
-            clusterCommunicator.unicast(clusterMessage, deviceNode);
+            clusterCommunicator.unicast(deviceInjectedEvent, DEVICE_INJECTED, SERIALIZER::encode, deviceNode);
             /* error log:
             /* error log:
             log.warn("Failed to process injected device id: {} desc: {} " +
             log.warn("Failed to process injected device id: {} desc: {} " +
                             "(cluster messaging failed: {})",
                             "(cluster messaging failed: {})",
@@ -555,11 +553,9 @@ public class GossipDeviceStore
             }
             }
 
 
             PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions);
             PortInjectedEvent portInjectedEvent = new PortInjectedEvent(providerId, deviceId, portDescriptions);
-            ClusterMessage clusterMessage = new ClusterMessage(
-                    localNode, PORT_INJECTED, SERIALIZER.encode(portInjectedEvent));
 
 
             //TODO check unicast return value
             //TODO check unicast return value
-            clusterCommunicator.unicast(clusterMessage, deviceNode);
+            clusterCommunicator.unicast(portInjectedEvent, PORT_INJECTED, SERIALIZER::encode, deviceNode);
             /* error log:
             /* error log:
             log.warn("Failed to process injected ports of device id: {} " +
             log.warn("Failed to process injected ports of device id: {} " +
                             "(cluster messaging failed: {})",
                             "(cluster messaging failed: {})",
@@ -867,13 +863,8 @@ public class GossipDeviceStore
             log.debug("{} has control of {}, forwarding remove request",
             log.debug("{} has control of {}, forwarding remove request",
                      master, deviceId);
                      master, deviceId);
 
 
-             ClusterMessage message = new ClusterMessage(
-                     myId,
-                     DEVICE_REMOVE_REQ,
-                     SERIALIZER.encode(deviceId));
-
             // TODO check unicast return value
             // TODO check unicast return value
-            clusterCommunicator.unicast(message, master);
+            clusterCommunicator.unicast(deviceId, DEVICE_REMOVE_REQ, SERIALIZER::encode, master);
              /* error log:
              /* error log:
              log.error("Failed to forward {} remove request to {}", deviceId, master, e);
              log.error("Failed to forward {} remove request to {}", deviceId, master, e);
              */
              */
@@ -1057,19 +1048,11 @@ public class GossipDeviceStore
     }
     }
 
 
     private void unicastMessage(NodeId recipient, MessageSubject subject, Object event) throws IOException {
     private void unicastMessage(NodeId recipient, MessageSubject subject, Object event) throws IOException {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                SERIALIZER.encode(event));
-        clusterCommunicator.unicast(message, recipient);
+        clusterCommunicator.unicast(event, subject, SERIALIZER::encode, recipient);
     }
     }
 
 
     private void broadcastMessage(MessageSubject subject, Object event) {
     private void broadcastMessage(MessageSubject subject, Object event) {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                SERIALIZER.encode(event));
-        clusterCommunicator.broadcast(message);
+        clusterCommunicator.broadcast(event, subject, SERIALIZER::encode);
     }
     }
 
 
     private void notifyPeers(InternalDeviceEvent event) {
     private void notifyPeers(InternalDeviceEvent event) {

@@ -510,11 +510,7 @@ public class EventuallyConsistentMapImpl<K, V>
     }
     }
 
 
     private boolean unicastMessage(NodeId peer, MessageSubject subject, Object event) {
     private boolean unicastMessage(NodeId peer, MessageSubject subject, Object event) {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                serializer.encode(event));
-        return clusterCommunicator.unicast(message, peer);
+        return clusterCommunicator.unicast(event, subject, serializer::encode, peer);
         // Note: we had this flipped before...
         // Note: we had this flipped before...
 //        communicationExecutor.execute(() -> clusterCommunicator.unicast(message, peer));
 //        communicationExecutor.execute(() -> clusterCommunicator.unicast(message, peer));
     }
     }

@@ -22,6 +22,7 @@ import com.google.common.collect.ImmutableList;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Iterables;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Maps;
 import com.google.common.collect.Sets;
 import com.google.common.collect.Sets;
+import com.google.common.util.concurrent.Futures;
 import com.hazelcast.core.IMap;
 import com.hazelcast.core.IMap;
 
 
 import org.apache.felix.scr.annotations.Activate;
 import org.apache.felix.scr.annotations.Activate;
@@ -35,6 +36,7 @@ import org.apache.felix.scr.annotations.Service;
 import org.onlab.util.BoundedThreadPool;
 import org.onlab.util.BoundedThreadPool;
 import org.onlab.util.KryoNamespace;
 import org.onlab.util.KryoNamespace;
 import org.onlab.util.NewConcurrentHashMap;
 import org.onlab.util.NewConcurrentHashMap;
+import org.onlab.util.Tools;
 import org.onosproject.cfg.ComponentConfigService;
 import org.onosproject.cfg.ComponentConfigService;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.NodeId;
 import org.onosproject.cluster.NodeId;
@@ -93,7 +95,6 @@ import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 import java.util.stream.Collectors;
 
 
 import static com.google.common.base.Preconditions.checkNotNull;
 import static com.google.common.base.Preconditions.checkNotNull;
@@ -360,21 +361,15 @@ public class DistributedFlowRuleStore
         log.trace("Forwarding getFlowEntry to {}, which is the primary (master) for device {}",
         log.trace("Forwarding getFlowEntry to {}, which is the primary (master) for device {}",
                   replicaInfo.master().orNull(), rule.deviceId());
                   replicaInfo.master().orNull(), rule.deviceId());
 
 
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
+        return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(rule,
                                     FlowStoreMessageSubjects.GET_FLOW_ENTRY,
                                     FlowStoreMessageSubjects.GET_FLOW_ENTRY,
-                SERIALIZER.encode(rule));
-
-        try {
-            Future<byte[]> responseFuture = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
-            return SERIALIZER.decode(responseFuture.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
-        } catch (IOException | TimeoutException | ExecutionException | InterruptedException e) {
-            log.warn("Unable to fetch flow store contents from {}", replicaInfo.master().get());
+                                    SERIALIZER::encode,
+                                    SERIALIZER::decode,
+                                    replicaInfo.master().get()),
+                               FLOW_RULE_STORE_TIMEOUT_MILLIS,
+                               TimeUnit.MILLISECONDS,
+                               null);
     }
     }
-        return null;
-    }
-
-
 
 
     @Override
     @Override
     public Iterable<FlowEntry> getFlowEntries(DeviceId deviceId) {
     public Iterable<FlowEntry> getFlowEntries(DeviceId deviceId) {
@@ -393,21 +388,15 @@ public class DistributedFlowRuleStore
         log.trace("Forwarding getFlowEntries to {}, which is the primary (master) for device {}",
         log.trace("Forwarding getFlowEntries to {}, which is the primary (master) for device {}",
                   replicaInfo.master().orNull(), deviceId);
                   replicaInfo.master().orNull(), deviceId);
 
 
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                GET_DEVICE_FLOW_ENTRIES,
-                SERIALIZER.encode(deviceId));
-
-        try {
-            Future<byte[]> responseFuture = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
-            return SERIALIZER.decode(responseFuture.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
-        } catch (IOException | TimeoutException | ExecutionException | InterruptedException e) {
-            log.warn("Unable to fetch flow store contents from {}", replicaInfo.master().get());
+        return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(deviceId,
+                                    FlowStoreMessageSubjects.GET_DEVICE_FLOW_ENTRIES,
+                                    SERIALIZER::encode,
+                                    SERIALIZER::decode,
+                                    replicaInfo.master().get()),
+                               FLOW_RULE_STORE_TIMEOUT_MILLIS,
+                               TimeUnit.MILLISECONDS,
+                               Collections.emptyList());
     }
     }
-        return Collections.emptyList();
-    }
-
-
 
 
     @Override
     @Override
     public void storeFlowRule(FlowRule rule) {
     public void storeFlowRule(FlowRule rule) {
@@ -453,14 +442,10 @@ public class DistributedFlowRuleStore
         log.trace("Forwarding storeBatch to {}, which is the primary (master) for device {}",
         log.trace("Forwarding storeBatch to {}, which is the primary (master) for device {}",
                   replicaInfo.master().orNull(), deviceId);
                   replicaInfo.master().orNull(), deviceId);
 
 
-        ClusterMessage message = new ClusterMessage(
-                local,
-                APPLY_BATCH_FLOWS,
-                SERIALIZER.encode(operation));
-
-
-        if (!clusterCommunicator.unicast(message, replicaInfo.master().get())) {
-            log.warn("Failed to storeBatch: {} to {}", message, replicaInfo.master());
+        if (!clusterCommunicator.unicast(operation,
+                APPLY_BATCH_FLOWS, SERIALIZER::encode,
+                replicaInfo.master().get())) {
+            log.warn("Failed to storeBatch: {} to {}", operation, replicaInfo.master());
 
 
             Set<FlowRule> allFailures = operation.getOperations().stream()
             Set<FlowRule> allFailures = operation.getOperations().stream()
                     .map(op -> op.target())
                     .map(op -> op.target())
@@ -612,18 +597,15 @@ public class DistributedFlowRuleStore
         log.trace("Forwarding removeFlowRule to {}, which is the primary (master) for device {}",
         log.trace("Forwarding removeFlowRule to {}, which is the primary (master) for device {}",
                   replicaInfo.master().orNull(), deviceId);
                   replicaInfo.master().orNull(), deviceId);
 
 
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
+        return Futures.get(clusterCommunicator.sendAndReceive(
+                               rule,
                                REMOVE_FLOW_ENTRY,
                                REMOVE_FLOW_ENTRY,
-                SERIALIZER.encode(rule));
-
-        try {
-            Future<byte[]> responseFuture = clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
-            return SERIALIZER.decode(responseFuture.get(FLOW_RULE_STORE_TIMEOUT_MILLIS, TimeUnit.MILLISECONDS));
-        } catch (IOException | TimeoutException | ExecutionException | InterruptedException e) {
-            // TODO: Retry against latest master or throw a FlowStoreException
-            throw new RuntimeException(e);
-        }
+                               SERIALIZER::encode,
+                               SERIALIZER::decode,
+                               replicaInfo.master().get()),
+                           FLOW_RULE_STORE_TIMEOUT_MILLIS,
+                           TimeUnit.MILLISECONDS,
+                           RuntimeException.class);
     }
     }
 
 
     private FlowRuleEvent removeFlowRuleInternal(FlowEntry rule) {
     private FlowRuleEvent removeFlowRuleInternal(FlowEntry rule) {
@@ -649,12 +631,8 @@ public class DistributedFlowRuleStore
         if (nodeId == null) {
         if (nodeId == null) {
             notifyDelegate(event);
             notifyDelegate(event);
         } else {
         } else {
-            ClusterMessage message = new ClusterMessage(
-                    clusterService.getLocalNode().id(),
-                    REMOTE_APPLY_COMPLETED,
-                    SERIALIZER.encode(event));
             // TODO check unicast return value
             // TODO check unicast return value
-            clusterCommunicator.unicast(message, nodeId);
+            clusterCommunicator.unicast(event, REMOTE_APPLY_COMPLETED, SERIALIZER::encode, nodeId);
             //error log: log.warn("Failed to respond to peer for batch operation result");
             //error log: log.warn("Failed to respond to peer for batch operation result");
         }
         }
     }
     }

@@ -20,6 +20,7 @@ import com.google.common.cache.CacheBuilder;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.Futures;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.ListenableFuture;
 import com.google.common.util.concurrent.SettableFuture;
 import com.google.common.util.concurrent.SettableFuture;
+
 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;
@@ -45,7 +46,6 @@ import org.onosproject.store.cluster.messaging.ClusterMessageHandler;
 import org.onosproject.store.flow.ReplicaInfo;
 import org.onosproject.store.flow.ReplicaInfo;
 import org.onosproject.store.flow.ReplicaInfoEventListener;
 import org.onosproject.store.flow.ReplicaInfoEventListener;
 import org.onosproject.store.flow.ReplicaInfoService;
 import org.onosproject.store.flow.ReplicaInfoService;
-import org.onosproject.store.serializers.DecodeTo;
 import org.onosproject.store.serializers.KryoSerializer;
 import org.onosproject.store.serializers.KryoSerializer;
 import org.onosproject.store.serializers.StoreSerializer;
 import org.onosproject.store.serializers.StoreSerializer;
 import org.onosproject.store.serializers.impl.DistributedStoreSerializers;
 import org.onosproject.store.serializers.impl.DistributedStoreSerializers;
@@ -199,18 +199,12 @@ public class DefaultFlowRuleExtRouter
         log.trace("Forwarding storeBatch to {}, which is the primary (master) for device {}",
         log.trace("Forwarding storeBatch to {}, which is the primary (master) for device {}",
                   replicaInfo.master().orNull(), deviceId);
                   replicaInfo.master().orNull(), deviceId);
 
 
-        ClusterMessage message = new ClusterMessage(clusterService
-                    .getLocalNode().id(), APPLY_EXTEND_FLOWS, SERIALIZER.encode(batchOperation));
-
-        try {
-            ListenableFuture<byte[]> responseFuture = clusterCommunicator
-                    .sendAndReceive(message, replicaInfo.master().get());
-            // here should add another decode process
-            return Futures.transform(responseFuture,
-                                     new DecodeTo<FlowExtCompletedOperation>(SERIALIZER));
-        } catch (IOException e) {
-            return Futures.immediateFailedFuture(e);
-        }
+        return clusterCommunicator.sendAndReceive(
+                batchOperation,
+                APPLY_EXTEND_FLOWS,
+                SERIALIZER::encode,
+                SERIALIZER::decode,
+                replicaInfo.master().get());
     }
     }
 
 
     /**
     /**

@@ -382,17 +382,13 @@ public class DistributedGroupStore
             GroupStoreMessage groupOp = GroupStoreMessage.
             GroupStoreMessage groupOp = GroupStoreMessage.
                     createGroupAddRequestMsg(groupDesc.deviceId(),
                     createGroupAddRequestMsg(groupDesc.deviceId(),
                                              groupDesc);
                                              groupDesc);
-            ClusterMessage message = new ClusterMessage(
-                                    clusterService.getLocalNode().id(),
-                                    GroupStoreMessageSubjects.
-                                    REMOTE_GROUP_OP_REQUEST,
-                                    kryoBuilder.build().serialize(groupOp));
-            if (!clusterCommunicator.unicast(message,
-                                             mastershipService.
-                                             getMasterFor(
-                                                groupDesc.deviceId()))) {
+
+            if (!clusterCommunicator.unicast(groupOp,
+                    GroupStoreMessageSubjects.REMOTE_GROUP_OP_REQUEST,
+                    m -> kryoBuilder.build().serialize(m),
+                    mastershipService.getMasterFor(groupDesc.deviceId()))) {
                 log.warn("Failed to send request to master: {} to {}",
                 log.warn("Failed to send request to master: {} to {}",
-                         message,
+                         groupOp,
                          mastershipService.getMasterFor(groupDesc.deviceId()));
                          mastershipService.getMasterFor(groupDesc.deviceId()));
                 //TODO: Send Group operation failure event
                 //TODO: Send Group operation failure event
             }
             }
@@ -472,16 +468,13 @@ public class DistributedGroupStore
                                                 type,
                                                 type,
                                                 newBuckets,
                                                 newBuckets,
                                                 newAppCookie);
                                                 newAppCookie);
-            ClusterMessage message =
-                    new ClusterMessage(clusterService.getLocalNode().id(),
-                                       GroupStoreMessageSubjects.
-                                              REMOTE_GROUP_OP_REQUEST,
-                                       kryoBuilder.build().serialize(groupOp));
-            if (!clusterCommunicator.unicast(message,
-                                             mastershipService.
-                                             getMasterFor(deviceId))) {
+
+            if (!clusterCommunicator.unicast(groupOp,
+                        GroupStoreMessageSubjects.REMOTE_GROUP_OP_REQUEST,
+                        m -> kryoBuilder.build().serialize(m),
+                        mastershipService.getMasterFor(deviceId))) {
                 log.warn("Failed to send request to master: {} to {}",
                 log.warn("Failed to send request to master: {} to {}",
-                         message,
+                         groupOp,
                          mastershipService.getMasterFor(deviceId));
                          mastershipService.getMasterFor(deviceId));
                 //TODO: Send Group operation failure event
                 //TODO: Send Group operation failure event
             }
             }
@@ -584,16 +577,13 @@ public class DistributedGroupStore
             GroupStoreMessage groupOp = GroupStoreMessage.
             GroupStoreMessage groupOp = GroupStoreMessage.
                     createGroupDeleteRequestMsg(deviceId,
                     createGroupDeleteRequestMsg(deviceId,
                                                 appCookie);
                                                 appCookie);
-            ClusterMessage message =
-                    new ClusterMessage(clusterService.getLocalNode().id(),
-                                       GroupStoreMessageSubjects.
-                                              REMOTE_GROUP_OP_REQUEST,
-                                       kryoBuilder.build().serialize(groupOp));
-            if (!clusterCommunicator.unicast(message,
-                                             mastershipService.
-                                             getMasterFor(deviceId))) {
+
+            if (!clusterCommunicator.unicast(groupOp,
+                    GroupStoreMessageSubjects.REMOTE_GROUP_OP_REQUEST,
+                    m -> kryoBuilder.build().serialize(m),
+                    mastershipService.getMasterFor(deviceId))) {
                 log.warn("Failed to send request to master: {} to {}",
                 log.warn("Failed to send request to master: {} to {}",
-                         message,
+                         groupOp,
                          mastershipService.getMasterFor(deviceId));
                          mastershipService.getMasterFor(deviceId));
                 //TODO: Send Group operation failure event
                 //TODO: Send Group operation failure event
             }
             }

@@ -477,21 +477,13 @@ public class GossipHostStore
     }
     }
 
 
     private void broadcastMessage(MessageSubject subject, Object event) {
     private void broadcastMessage(MessageSubject subject, Object event) {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                SERIALIZER.encode(event));
-        clusterCommunicator.broadcast(message);
+        clusterCommunicator.broadcast(event, subject, SERIALIZER::encode);
     }
     }
 
 
     private void unicastMessage(NodeId peer,
     private void unicastMessage(NodeId peer,
                                 MessageSubject subject,
                                 MessageSubject subject,
                                 Object event) throws IOException {
                                 Object event) throws IOException {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                SERIALIZER.encode(event));
-        clusterCommunicator.unicast(message, peer);
+        clusterCommunicator.unicast(event, subject, SERIALIZER::encode, peer);
     }
     }
 
 
     private void notifyDelegateIfNotNull(HostEvent event) {
     private void notifyDelegateIfNotNull(HostEvent event) {

@@ -334,17 +334,12 @@ public class GossipLinkStore
 
 
 
 
             LinkInjectedEvent linkInjectedEvent = new LinkInjectedEvent(providerId, linkDescription);
             LinkInjectedEvent linkInjectedEvent = new LinkInjectedEvent(providerId, linkDescription);
-            ClusterMessage linkInjectedMessage = new ClusterMessage(localNode,
-                    GossipLinkStoreMessageSubjects.LINK_INJECTED, SERIALIZER.encode(linkInjectedEvent));
 
 
             // TODO check unicast return value
             // TODO check unicast return value
-            clusterCommunicator.unicast(linkInjectedMessage, dstNode);
-            /* error log:
-            log.warn("Failed to process link update between src: {} and dst: {} " +
-                            "(cluster messaging failed: {})",
-                    linkDescription.src(), linkDescription.dst(), e);
-            */
-
+            clusterCommunicator.unicast(linkInjectedEvent,
+                    GossipLinkStoreMessageSubjects.LINK_INJECTED,
+                    SERIALIZER::encode,
+                    dstNode);
         }
         }
 
 
         return linkEvent;
         return linkEvent;
@@ -653,19 +648,11 @@ public class GossipLinkStore
     }
     }
 
 
     private void broadcastMessage(MessageSubject subject, Object event) {
     private void broadcastMessage(MessageSubject subject, Object event) {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                SERIALIZER.encode(event));
-        clusterCommunicator.broadcast(message);
+        clusterCommunicator.broadcast(event, subject, SERIALIZER::encode);
     }
     }
 
 
     private void unicastMessage(NodeId recipient, MessageSubject subject, Object event) throws IOException {
     private void unicastMessage(NodeId recipient, MessageSubject subject, Object event) throws IOException {
-        ClusterMessage message = new ClusterMessage(
-                clusterService.getLocalNode().id(),
-                subject,
-                SERIALIZER.encode(event));
-        clusterCommunicator.unicast(message, recipient);
+        clusterCommunicator.unicast(event, subject, SERIALIZER::encode, recipient);
     }
     }
 
 
     private void notifyPeers(InternalLinkEvent event) {
     private void notifyPeers(InternalLinkEvent event) {

@@ -181,20 +181,14 @@ public class ConsistentDeviceMastershipStore
             } else {
             } else {
                 return MastershipRole.NONE;
                 return MastershipRole.NONE;
             }
             }
-        } else {
-            try {
+        }
         MastershipRole role = complete(clusterCommunicator.sendAndReceive(
         MastershipRole role = complete(clusterCommunicator.sendAndReceive(
-                                                        new ClusterMessage(
-                                                                    localNodeId,
+                deviceId,
                 ROLE_QUERY_SUBJECT,
                 ROLE_QUERY_SUBJECT,
-                                                                    SERIALIZER.encode(deviceId)),
+                SERIALIZER::encode,
+                SERIALIZER::decode,
                 nodeId));
                 nodeId));
         return role == null ? MastershipRole.NONE : role;
         return role == null ? MastershipRole.NONE : role;
-            } catch (IOException e) {
-                log.warn("Failed to query {} for {}'s role. Defaulting to NONE", nodeId, deviceId, e);
-                return MastershipRole.NONE;
-            }
-        }
     }
     }
 
 
     @Override
     @Override
@@ -276,17 +270,12 @@ public class ConsistentDeviceMastershipStore
         if (!nodeId.equals(localNodeId)) {
         if (!nodeId.equals(localNodeId)) {
             log.debug("Forwarding request to relinquish "
             log.debug("Forwarding request to relinquish "
                     + "role for device {} to {}", deviceId, nodeId);
                     + "role for device {} to {}", deviceId, nodeId);
-            try {
             return complete(clusterCommunicator.sendAndReceive(
             return complete(clusterCommunicator.sendAndReceive(
-                                                        new ClusterMessage(
-                                                                    localNodeId,
+                    deviceId,
                     ROLE_RELINQUISH_SUBJECT,
                     ROLE_RELINQUISH_SUBJECT,
-                                                                    SERIALIZER.encode(deviceId)),
+                    SERIALIZER::encode,
+                    SERIALIZER::decode,
                     nodeId));
                     nodeId));
-            } catch (IOException e) {
-                log.warn("Failed to send a request to relinquish role for {} to {}", deviceId, nodeId, e);
-                return null;
-            }
         }
         }
 
 
         // Check if this node is can be managed by this node.
         // Check if this node is can be managed by this node.

@@ -131,9 +131,7 @@ public class DistributedPacketStore
         }
         }
 
 
         // TODO check unicast return value
         // TODO check unicast return value
-        communicationService.unicast(new ClusterMessage(myId, PACKET_OUT_SUBJECT,
-                                                        SERIALIZER.encode(packet)),
-                                     master);
+        communicationService.unicast(packet, PACKET_OUT_SUBJECT, SERIALIZER::encode, master);
         // error log: log.warn("Failed to send packet-out to {}", master);
         // error log: log.warn("Failed to send packet-out to {}", master);
     }
     }
 
 

@@ -16,6 +16,7 @@
 package org.onosproject.store.statistic.impl;
 package org.onosproject.store.statistic.impl;
 
 
 import com.google.common.collect.Sets;
 import com.google.common.collect.Sets;
+
 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;
@@ -23,6 +24,7 @@ import org.apache.felix.scr.annotations.Reference;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.ReferenceCardinality;
 import org.apache.felix.scr.annotations.Service;
 import org.apache.felix.scr.annotations.Service;
 import org.onlab.util.KryoNamespace;
 import org.onlab.util.KryoNamespace;
+import org.onlab.util.Tools;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.cluster.ClusterService;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.ConnectPoint;
 import org.onosproject.net.DeviceId;
 import org.onosproject.net.DeviceId;
@@ -47,12 +49,9 @@ import java.util.HashSet;
 import java.util.Map;
 import java.util.Map;
 import java.util.Set;
 import java.util.Set;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
-import java.util.concurrent.ExecutionException;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Executors;
-import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeUnit;
-import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicInteger;
 import java.util.concurrent.atomic.AtomicInteger;
 
 
 import static org.onlab.util.Tools.groupedThreads;
 import static org.onlab.util.Tools.groupedThreads;
@@ -218,20 +217,15 @@ public class DistributedStatisticStore implements StatisticStore {
         if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
         if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
             return getCurrentStatisticInternal(connectPoint);
             return getCurrentStatisticInternal(connectPoint);
         } else {
         } else {
-            ClusterMessage message = new ClusterMessage(
-                    clusterService.getLocalNode().id(),
+            return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(
+                                        connectPoint,
                                         GET_CURRENT,
                                         GET_CURRENT,
-                    SERIALIZER.encode(connectPoint));
-
-            try {
-                Future<byte[]> response =
-                        clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
-                return SERIALIZER.decode(response.get(STATISTIC_STORE_TIMEOUT_MILLIS,
-                                                      TimeUnit.MILLISECONDS));
-            } catch (IOException | TimeoutException | ExecutionException | InterruptedException e) {
-                log.warn("Unable to communicate with peer {}", replicaInfo.master().get());
-                return Collections.emptySet();
-            }
+                                        SERIALIZER::encode,
+                                        SERIALIZER::decode,
+                                        replicaInfo.master().get()),
+                                   STATISTIC_STORE_TIMEOUT_MILLIS,
+                                   TimeUnit.MILLISECONDS,
+                                   Collections.emptySet());
         }
         }
 
 
     }
     }
@@ -251,24 +245,18 @@ public class DistributedStatisticStore implements StatisticStore {
         if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
         if (replicaInfo.master().get().equals(clusterService.getLocalNode().id())) {
             return getPreviousStatisticInternal(connectPoint);
             return getPreviousStatisticInternal(connectPoint);
         } else {
         } else {
-            ClusterMessage message = new ClusterMessage(
-                    clusterService.getLocalNode().id(),
+            return Tools.futureGetOrElse(clusterCommunicator.sendAndReceive(
+                                        connectPoint,
                                         GET_PREVIOUS,
                                         GET_PREVIOUS,
-                    SERIALIZER.encode(connectPoint));
-
-            try {
-                Future<byte[]> response =
-                        clusterCommunicator.sendAndReceive(message, replicaInfo.master().get());
-                return SERIALIZER.decode(response.get(STATISTIC_STORE_TIMEOUT_MILLIS,
-                                                      TimeUnit.MILLISECONDS));
-            } catch (IOException | TimeoutException | ExecutionException | InterruptedException e) {
-                log.warn("Unable to communicate with peer {}", replicaInfo.master().get());
-                return Collections.emptySet();
+                                        SERIALIZER::encode,
+                                        SERIALIZER::decode,
+                                        replicaInfo.master().get()),
+                                   STATISTIC_STORE_TIMEOUT_MILLIS,
+                                   TimeUnit.MILLISECONDS,
+                                   Collections.emptySet());
         }
         }
     }
     }
 
 
-    }
-
     private synchronized Set<FlowEntry> getPreviousStatisticInternal(ConnectPoint connectPoint) {
     private synchronized Set<FlowEntry> getPreviousStatisticInternal(ConnectPoint connectPoint) {
         return previous.get(connectPoint);
         return previous.get(connectPoint);
     }
     }

-/*
- * Copyright 2014 Open Networking Laboratory
- *
- * Licensed under the Apache License, Version 2.0 (the "License");
- * you may not use this file except in compliance with the License.
- * You may obtain a copy of the License at
- *
- *     http://www.apache.org/licenses/LICENSE-2.0
- *
- * Unless required by applicable law or agreed to in writing, software
- * distributed under the License is distributed on an "AS IS" BASIS,
- * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
- * See the License for the specific language governing permissions and
- * limitations under the License.
- */
-package org.onosproject.store.serializers;
-
-import static com.google.common.base.Preconditions.checkNotNull;
-
-import com.google.common.base.Function;
-
-/**
- * Function to convert byte[] into {@code T}.
- *
- * @param <T> Type after decoding
- */
-public final class DecodeTo<T> implements Function<byte[], T> {
-
-    private StoreSerializer serializer;
-
-    public DecodeTo(StoreSerializer serializer) {
-        this.serializer = checkNotNull(serializer);
-    }
-
-    @Override
-    public T apply(byte[] input) {
-        return serializer.decode(input);
-    }
-}

@@ -15,10 +15,10 @@
  */
  */
 package org.onlab.util;
 package org.onlab.util;
 
 
-import com.google.common.base.Strings;
-import com.google.common.primitives.UnsignedLongs;
-import com.google.common.util.concurrent.ThreadFactoryBuilder;
-import org.slf4j.Logger;
+import static java.nio.file.Files.delete;
+import static java.nio.file.Files.walkFileTree;
+import static org.onlab.util.GroupedThreadFactory.groupedThreadFactory;
+import static org.slf4j.LoggerFactory.getLogger;
 
 
 import java.io.BufferedReader;
 import java.io.BufferedReader;
 import java.io.File;
 import java.io.File;
@@ -37,12 +37,17 @@ import java.util.ArrayList;
 import java.util.Collection;
 import java.util.Collection;
 import java.util.Dictionary;
 import java.util.Dictionary;
 import java.util.List;
 import java.util.List;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.Future;
 import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadFactory;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.TimeoutException;
 
 
-import static java.nio.file.Files.delete;
-import static java.nio.file.Files.walkFileTree;
-import static org.onlab.util.GroupedThreadFactory.groupedThreadFactory;
-import static org.slf4j.LoggerFactory.getLogger;
+import org.slf4j.Logger;
+
+import com.google.common.base.Strings;
+import com.google.common.primitives.UnsignedLongs;
+import com.google.common.util.concurrent.ThreadFactoryBuilder;
 
 
 /**
 /**
  * Miscellaneous utility methods.
  * Miscellaneous utility methods.
@@ -324,6 +329,51 @@ public abstract class Tools {
                                          dst.getAbsolutePath()));
                                          dst.getAbsolutePath()));
     }
     }
 
 
+    /**
+     * Returns the future value when complete or if future
+     * completes exceptionally returns the defaultValue.
+     * @param future future
+     * @param defaultValue default value
+     * @param <T> future value type
+     * @return future value when complete or if future
+     * completes exceptionally returns the defaultValue.
+     */
+    public static <T> T futureGetOrElse(Future<T> future, T defaultValue) {
+        try {
+            return future.get();
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return defaultValue;
+        } catch (ExecutionException e) {
+            return defaultValue;
+        }
+    }
+
+    /**
+     * Returns the future value when complete or if future
+     * completes exceptionally returns the defaultValue.
+     * @param future future
+     * @param timeout time to wait for successful completion
+     * @param timeUnit time unit
+     * @param defaultValue default value
+     * @param <T> future value type
+     * @return future value when complete or if future
+     * completes exceptionally returns the defaultValue.
+     */
+    public static <T> T futureGetOrElse(Future<T> future,
+                                        long timeout,
+                                        TimeUnit timeUnit,
+                                        T defaultValue) {
+        try {
+            return future.get(timeout, timeUnit);
+        } catch (InterruptedException e) {
+            Thread.currentThread().interrupt();
+            return defaultValue;
+        } catch (ExecutionException | TimeoutException e) {
+            return defaultValue;
+        }
+    }
+
     // Auxiliary path visitor for recursive directory structure copying.
     // Auxiliary path visitor for recursive directory structure copying.
     private static class DirectoryCopier extends SimpleFileVisitor<Path> {
     private static class DirectoryCopier extends SimpleFileVisitor<Path> {
         private Path src;
         private Path src;

@@ -16,10 +16,9 @@
 package org.onlab.netty;
 package org.onlab.netty;
 
 
 import java.io.IOException;
 import java.io.IOException;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
 
 
-import com.google.common.util.concurrent.ListenableFuture;
-
 /**
 /**
  * Interface for low level messaging primitives.
  * Interface for low level messaging primitives.
  */
  */
@@ -40,9 +39,8 @@ public interface MessagingService {
      * @param type type of message.
      * @param type type of message.
      * @param payload message payload.
      * @param payload message payload.
      * @return a response future
      * @return a response future
-     * @throws IOException when I/O exception of some sort has occurred
      */
      */
-    public ListenableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) throws IOException;
+    public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload);
 
 
     /**
     /**
      * Registers a new message handler for message type.
      * Registers a new message handler for message type.

@@ -39,6 +39,7 @@ import io.netty.channel.socket.nio.NioSocketChannel;
 import java.io.IOException;
 import java.io.IOException;
 import java.net.InetAddress;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.net.UnknownHostException;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentHashMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ConcurrentMap;
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.ExecutorService;
@@ -56,8 +57,6 @@ import com.google.common.cache.Cache;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.CacheBuilder;
 import com.google.common.cache.RemovalListener;
 import com.google.common.cache.RemovalListener;
 import com.google.common.cache.RemovalNotification;
 import com.google.common.cache.RemovalNotification;
-import com.google.common.util.concurrent.ListenableFuture;
-import com.google.common.util.concurrent.SettableFuture;
 
 
 /**
 /**
  * A Netty based implementation of MessagingService.
  * A Netty based implementation of MessagingService.
@@ -69,14 +68,14 @@ public class NettyMessagingService implements MessagingService {
     private final Endpoint localEp;
     private final Endpoint localEp;
     private final ConcurrentMap<String, MessageHandler> handlers = new ConcurrentHashMap<>();
     private final ConcurrentMap<String, MessageHandler> handlers = new ConcurrentHashMap<>();
     private final AtomicLong messageIdGenerator = new AtomicLong(0);
     private final AtomicLong messageIdGenerator = new AtomicLong(0);
-    private final Cache<Long, SettableFuture<byte[]>> responseFutures = CacheBuilder.newBuilder()
+    private final Cache<Long, CompletableFuture<byte[]>> responseFutures = CacheBuilder.newBuilder()
             .maximumSize(100000)
             .maximumSize(100000)
             .expireAfterWrite(10, TimeUnit.SECONDS)
             .expireAfterWrite(10, TimeUnit.SECONDS)
-            .removalListener(new RemovalListener<Long, SettableFuture<byte[]>>() {
+            .removalListener(new RemovalListener<Long, CompletableFuture<byte[]>>() {
                 @Override
                 @Override
-                public void onRemoval(RemovalNotification<Long, SettableFuture<byte[]>> entry) {
+                public void onRemoval(RemovalNotification<Long, CompletableFuture<byte[]>> entry) {
                     if (entry.wasEvicted()) {
                     if (entry.wasEvicted()) {
-                        entry.getValue().setException(new TimeoutException("Timedout waiting for reply"));
+                        entry.getValue().completeExceptionally(new TimeoutException("Timedout waiting for reply"));
                     }
                     }
                 }
                 }
             })
             })
@@ -178,11 +177,10 @@ public class NettyMessagingService implements MessagingService {
     }
     }
 
 
     @Override
     @Override
-    public ListenableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload)
-            throws IOException {
-        SettableFuture<byte[]> futureResponse = SettableFuture.create();
+    public CompletableFuture<byte[]> sendAndReceive(Endpoint ep, String type, byte[] payload) {
+        CompletableFuture<byte[]> response = new CompletableFuture<>();
         Long messageId = messageIdGenerator.incrementAndGet();
         Long messageId = messageIdGenerator.incrementAndGet();
-        responseFutures.put(messageId, futureResponse);
+        responseFutures.put(messageId, response);
         InternalMessage message = new InternalMessage.Builder(this)
         InternalMessage message = new InternalMessage.Builder(this)
             .withId(messageId)
             .withId(messageId)
             .withSender(localEp)
             .withSender(localEp)
@@ -193,9 +191,9 @@ public class NettyMessagingService implements MessagingService {
             sendAsync(ep, message);
             sendAsync(ep, message);
         } catch (Exception e) {
         } catch (Exception e) {
             responseFutures.invalidate(messageId);
             responseFutures.invalidate(messageId);
-            throw e;
+            response.completeExceptionally(e);
         }
         }
-        return futureResponse;
+        return response;
     }
     }
 
 
     @Override
     @Override
@@ -333,10 +331,10 @@ public class NettyMessagingService implements MessagingService {
         String type = message.type();
         String type = message.type();
         if (InternalMessage.REPLY_MESSAGE_TYPE.equals(type)) {
         if (InternalMessage.REPLY_MESSAGE_TYPE.equals(type)) {
             try {
             try {
-                SettableFuture<byte[]> futureResponse =
+                CompletableFuture<byte[]> futureResponse =
                     NettyMessagingService.this.responseFutures.getIfPresent(message.id());
                     NettyMessagingService.this.responseFutures.getIfPresent(message.id());
                 if (futureResponse != null) {
                 if (futureResponse != null) {
-                    futureResponse.set(message.payload());
+                    futureResponse.complete(message.payload());
                 } else {
                 } else {
                     log.warn("Received a reply for message id:[{}]. "
                     log.warn("Received a reply for message id:[{}]. "
                             + " from {}. But was unable to locate the"
                             + " from {}. But was unable to locate the"