ClusterMessagingProtocolServer: start listening at correct timing

Change-Id: Ie8ed1894ae16c41242aee861440174f011dd689b

@@ -73,4 +73,12 @@ public interface ClusterCommunicationService {
      * @param subscriber message subscriber
      * @param subscriber message subscriber
      */
      */
     void addSubscriber(MessageSubject subject, ClusterMessageHandler subscriber);
     void addSubscriber(MessageSubject subject, ClusterMessageHandler subscriber);
+
+    /**
+     * Removes a subscriber for the specified message subject.
+     *
+     * @param subject    message subject
+     */
+    void removeSubscriber(MessageSubject subject);
+
 }
 }

@@ -166,10 +166,15 @@ public class ClusterCommunicationManager
 
 
     @Override
     @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));
     }
     }
 
 
+    @Override
+    public void removeSubscriber(MessageSubject subject) {
+        messagingService.unregisterHandler(subject.value());
+    }
+
     private final class InternalClusterMessageHandler implements MessageHandler {
     private final class InternalClusterMessageHandler implements MessageHandler {
 
 
         private final ClusterMessageHandler handler;
         private final ClusterMessageHandler handler;

@@ -27,18 +27,12 @@ import org.slf4j.Logger;
 public class ClusterMessagingProtocolServer implements ProtocolServer {
 public class ClusterMessagingProtocolServer implements ProtocolServer {
 
 
     private final Logger log = getLogger(getClass());
     private final Logger log = getLogger(getClass());
-    private RequestHandler handler;
+    private volatile RequestHandler handler;
+    private ClusterCommunicationService clusterCommunicator;
 
 
     public ClusterMessagingProtocolServer(ClusterCommunicationService clusterCommunicator) {
     public ClusterMessagingProtocolServer(ClusterCommunicationService clusterCommunicator) {
+        this.clusterCommunicator = clusterCommunicator;
 
 
-        clusterCommunicator.addSubscriber(
-                ClusterMessagingProtocol.COPYCAT_PING, new CopycatMessageHandler<PingRequest>());
-        clusterCommunicator.addSubscriber(
-                ClusterMessagingProtocol.COPYCAT_SYNC, new CopycatMessageHandler<SyncRequest>());
-        clusterCommunicator.addSubscriber(
-                ClusterMessagingProtocol.COPYCAT_POLL, new CopycatMessageHandler<PollRequest>());
-        clusterCommunicator.addSubscriber(
-                ClusterMessagingProtocol.COPYCAT_SUBMIT, new CopycatMessageHandler<SubmitRequest>());
     }
     }
 
 
     @Override
     @Override
@@ -48,11 +42,23 @@ public class ClusterMessagingProtocolServer implements ProtocolServer {
 
 
     @Override
     @Override
     public CompletableFuture<Void> listen() {
     public CompletableFuture<Void> listen() {
+        clusterCommunicator.addSubscriber(ClusterMessagingProtocol.COPYCAT_PING,
+                                          new CopycatMessageHandler<PingRequest>());
+        clusterCommunicator.addSubscriber(ClusterMessagingProtocol.COPYCAT_SYNC,
+                                          new CopycatMessageHandler<SyncRequest>());
+        clusterCommunicator.addSubscriber(ClusterMessagingProtocol.COPYCAT_POLL,
+                                          new CopycatMessageHandler<PollRequest>());
+        clusterCommunicator.addSubscriber(ClusterMessagingProtocol.COPYCAT_SUBMIT,
+                                          new CopycatMessageHandler<SubmitRequest>());
         return CompletableFuture.completedFuture(null);
         return CompletableFuture.completedFuture(null);
     }
     }
 
 
     @Override
     @Override
     public CompletableFuture<Void> close() {
     public CompletableFuture<Void> close() {
+        clusterCommunicator.removeSubscriber(ClusterMessagingProtocol.COPYCAT_PING);
+        clusterCommunicator.removeSubscriber(ClusterMessagingProtocol.COPYCAT_SYNC);
+        clusterCommunicator.removeSubscriber(ClusterMessagingProtocol.COPYCAT_POLL);
+        clusterCommunicator.removeSubscriber(ClusterMessagingProtocol.COPYCAT_SUBMIT);
         return CompletableFuture.completedFuture(null);
         return CompletableFuture.completedFuture(null);
     }
     }