Separate ClusterCommunicationService from RaftClient with protocol API.

core/src/main/java/io/atomix/protocols/raft/client/RaftClient.java

@@ -15,9 +15,9 @@
  */
 package io.atomix.protocols.raft.client;
 
-import io.atomix.cluster.ClusterCommunicationService;
 import io.atomix.cluster.NodeId;
 import io.atomix.protocols.raft.client.impl.DefaultRaftClient;
+import io.atomix.protocols.raft.protocol.RaftClientProtocol;
 import io.atomix.protocols.raft.session.RaftSession;
 
 import java.util.Arrays;
@@ -81,9 +81,9 @@ static Builder builder(Collection<NodeId> cluster) {
     }
 
     /**
-     * Returns the client identifier.
+     * Returns the globally unique client identifier.
      *
-     * @return The client identifier.
+     * @return the globally unique client identifier
      */
     String id();
 
@@ -166,8 +166,7 @@ default CompletableFuture<RaftClient> connect(NodeId... members) {
     final class Builder implements io.atomix.util.Builder<RaftClient> {
         private final Collection<NodeId> cluster;
         private String clientId = UUID.randomUUID().toString();
-        private String clusterName = "raft";
-        private ClusterCommunicationService clusterCommunicator;
+        private RaftClientProtocol protocol;
         private int threadPoolSize = Runtime.getRuntime().availableProcessors();
 
         private Builder(Collection<NodeId> cluster) {
@@ -185,19 +184,19 @@ private Builder(Collection<NodeId> cluster) {
          * @throws NullPointerException if {@code clientId} is null
          */
         public Builder withClientId(String clientId) {
-            this.clientId = checkNotNull(clientId, "clientId");
+            this.clientId = checkNotNull(clientId, "clientId cannot be null");
             return this;
         }
 
         /**
-         * Sets the cluster name.
+         * Sets the client protocol.
          *
-         * @param clusterName the cluster name
+         * @param protocol the client protocol
          * @return the client builder
-         * @throws NullPointerException if the cluster name is null
+         * @throws NullPointerException if the protocol is null
          */
-        public Builder withClusterName(String clusterName) {
-            this.clusterName = checkNotNull(clusterName, "clusterName cannot be null");
+        public Builder withProtocol(RaftClientProtocol protocol) {
+            this.protocol = checkNotNull(protocol, "protocol cannot be null");
             return this;
         }
 
@@ -216,15 +215,8 @@ public Builder withThreadPoolSize(int threadPoolSize) {
 
         @Override
         public RaftClient build() {
-            checkNotNull(clusterName, "clusterName cannot be null");
             ScheduledExecutorService executor = Executors.newScheduledThreadPool(threadPoolSize);
-            return new DefaultRaftClient(
-                    clientId,
-                    clusterName,
-                    cluster,
-                    clusterCommunicator,
-                    executor);
+            return new DefaultRaftClient(clientId, cluster, protocol, executor);
         }
     }
-
 }

core/src/main/java/io/atomix/protocols/raft/client/impl/DefaultRaftClient.java

@@ -15,13 +15,13 @@
  */
 package io.atomix.protocols.raft.client.impl;
 
-import io.atomix.cluster.ClusterCommunicationService;
 import io.atomix.cluster.NodeId;
 import io.atomix.protocols.raft.client.RaftClient;
 import io.atomix.protocols.raft.client.RaftMetadataClient;
+import io.atomix.protocols.raft.protocol.RaftClientProtocol;
 import io.atomix.protocols.raft.session.RaftSession;
-import io.atomix.protocols.raft.session.impl.RaftSessionManager;
 import io.atomix.protocols.raft.session.impl.NodeSelectorManager;
+import io.atomix.protocols.raft.session.impl.RaftSessionManager;
 
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
@@ -36,33 +36,28 @@
  * @author <a href="http://github.com/kuujo>Jordan Halterman</a>
  */
 public class DefaultRaftClient implements RaftClient {
-    private static final String DEFAULT_HOST = "0.0.0.0";
-    private static final int DEFAULT_PORT = 8700;
-    private final String id;
+    private final String clientId;
     private final Collection<NodeId> cluster;
-    private final ClusterCommunicationService communicationService;
     private final ScheduledExecutorService threadPoolExecutor;
     private final RaftMetadataClient metadata;
     private final NodeSelectorManager selectorManager = new NodeSelectorManager();
     private final RaftSessionManager sessionManager;
 
     public DefaultRaftClient(
             String clientId,
-            String clusterName,
             Collection<NodeId> cluster,
-            ClusterCommunicationService communicationService,
+            RaftClientProtocol protocol,
             ScheduledExecutorService threadPoolExecutor) {
-        this.id = checkNotNull(clientId, "clientId cannot be null");
+        this.clientId = checkNotNull(clientId, "clientId cannot be null");
         this.cluster = checkNotNull(cluster, "cluster cannot be null");
-        this.communicationService = checkNotNull(communicationService, "communicationService cannot be null");
         this.threadPoolExecutor = checkNotNull(threadPoolExecutor, "threadPoolExecutor cannot be null");
-        this.metadata = new DefaultRaftMetadataClient(clientId, clusterName, communicationService, selectorManager);
-        this.sessionManager = new RaftSessionManager(clientId, clusterName, communicationService, selectorManager, threadPoolExecutor);
+        this.metadata = new DefaultRaftMetadataClient(clientId, protocol, selectorManager);
+        this.sessionManager = new RaftSessionManager(clientId, protocol, selectorManager, threadPoolExecutor);
     }
 
     @Override
     public String id() {
-        return id;
+        return clientId;
     }
 
     @Override
@@ -108,20 +103,10 @@ public synchronized CompletableFuture<Void> close() {
         return sessionManager.close();
     }
 
-    @Override
-    public int hashCode() {
-        return 23 + 37 * id.hashCode();
-    }
-
-    @Override
-    public boolean equals(Object object) {
-        return object instanceof DefaultRaftClient && ((DefaultRaftClient) object).id.equals(id);
-    }
-
     @Override
     public String toString() {
         return toStringHelper(this)
-                .add("id", id)
+                .add("id", clientId)
                 .toString();
     }
 
@@ -134,5 +119,4 @@ public RaftSession build() {
             return sessionManager.openSession(name, type, communicationStrategy, serializer, timeout).join();
         }
     }
-
 }

core/src/main/java/io/atomix/protocols/raft/client/impl/DefaultRaftMetadataClient.java

@@ -15,16 +15,16 @@
  */
 package io.atomix.protocols.raft.client.impl;
 
-import io.atomix.cluster.ClusterCommunicationService;
 import io.atomix.cluster.NodeId;
 import io.atomix.protocols.raft.client.CommunicationStrategies;
 import io.atomix.protocols.raft.client.RaftMetadataClient;
-import io.atomix.protocols.raft.session.impl.RaftClientConnection;
-import io.atomix.protocols.raft.session.impl.NodeSelectorManager;
-import io.atomix.protocols.raft.metadata.RaftSessionMetadata;
 import io.atomix.protocols.raft.protocol.MetadataRequest;
 import io.atomix.protocols.raft.protocol.MetadataResponse;
+import io.atomix.protocols.raft.protocol.RaftClientProtocol;
 import io.atomix.protocols.raft.protocol.RaftResponse;
+import io.atomix.protocols.raft.metadata.RaftSessionMetadata;
+import io.atomix.protocols.raft.session.impl.NodeSelectorManager;
+import io.atomix.protocols.raft.session.impl.RaftClientConnection;
 
 import java.util.Collection;
 import java.util.Set;
@@ -40,9 +40,9 @@ public class DefaultRaftMetadataClient implements RaftMetadataClient {
   private final NodeSelectorManager selectorManager;
   private final RaftClientConnection connection;
 
-  public DefaultRaftMetadataClient(String clientId, String clusterName, ClusterCommunicationService communicationService, NodeSelectorManager selectorManager) {
+  public DefaultRaftMetadataClient(String clientId, RaftClientProtocol protocol, NodeSelectorManager selectorManager) {
     this.selectorManager = checkNotNull(selectorManager, "selectorManager cannot be null");
-    this.connection = new RaftClientConnection(clientId, clusterName, communicationService, selectorManager.createSelector(CommunicationStrategies.LEADER));
+    this.connection = new RaftClientConnection(clientId, protocol.dispatcher(), selectorManager.createSelector(CommunicationStrategies.LEADER));
   }
 
   @Override
@@ -62,7 +62,7 @@ public Collection<NodeId> servers() {
    */
   private CompletableFuture<MetadataResponse> getMetadata() {
     CompletableFuture<MetadataResponse> future = new CompletableFuture<>();
-    connection.<MetadataRequest, MetadataResponse>sendAndReceive(MetadataRequest.builder().build()).whenComplete((response, error) -> {
+    connection.metadata(MetadataRequest.builder().build()).whenComplete((response, error) -> {
       if (error == null) {
         if (response.status() == RaftResponse.Status.OK) {
           future.complete(response);

core/src/main/java/io/atomix/protocols/raft/protocol/AbstractRaftResponse.java

@@ -28,83 +28,92 @@
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
 public abstract class AbstractRaftResponse implements RaftResponse {
-  protected final Status status;
-  protected final RaftError error;
+    protected final Status status;
+    protected final RaftError error;
 
-  protected AbstractRaftResponse(Status status, RaftError error) {
-    this.status = status;
-    this.error = error;
-  }
-
-  @Override
-  public Status status() {
-    return status;
-  }
-
-  @Override
-  public RaftError error() {
-    return error;
-  }
+    protected AbstractRaftResponse(Status status, RaftError error) {
+        this.status = status;
+        this.error = error;
+    }
 
-  @Override
-  public int hashCode() {
-    return Objects.hash(getClass(), status);
-  }
+    @Override
+    public Status status() {
+        return status;
+    }
 
-  @Override
-  public String toString() {
-    if (status == Status.OK) {
-      return toStringHelper(this)
-              .add("status", status)
-              .toString();
-    } else {
-      return toStringHelper(this)
-              .add("status", status)
-              .add("error", error)
-              .toString();
+    @Override
+    public RaftError error() {
+        return error;
     }
-  }
 
-  /**
-   * Abstract response builder.
-   *
-   * @param <T> The builder type.
-   * @param <U> The response type.
-   */
-  protected static abstract class Builder<T extends Builder<T, U>, U extends AbstractRaftResponse> implements RaftResponse.Builder<T, U> {
-    protected Status status;
-    protected RaftError error;
+    @Override
+    public int hashCode() {
+        return Objects.hash(getClass(), status);
+    }
 
     @Override
-    @SuppressWarnings("unchecked")
-    public T withStatus(Status status) {
-      this.status = checkNotNull(status, "status cannot be null");
-      return (T) this;
+    public boolean equals(Object object) {
+        if (object.getClass() == getClass()) {
+            AbstractRaftResponse response = (AbstractRaftResponse) object;
+            return response.status == status && Objects.equals(response.error, error);
+        }
+        return false;
     }
 
     @Override
-    @SuppressWarnings("unchecked")
-    public T withError(RaftError error) {
-      this.error = checkNotNull(error, "error cannot be null");
-      return (T) this;
+    public String toString() {
+        if (status == Status.OK) {
+            return toStringHelper(this)
+                    .add("status", status)
+                    .toString();
+        } else {
+            return toStringHelper(this)
+                    .add("status", status)
+                    .add("error", error)
+                    .toString();
+        }
     }
 
     /**
-     * Validates the builder.
+     * Abstract response builder.
+     *
+     * @param <T> The builder type.
+     * @param <U> The response type.
      */
-    protected void validate() {
-      checkNotNull(status, "status cannot be null");
-      if (status == Status.ERROR) {
-        checkNotNull(error, "error cannot be null");
-      }
-    }
+    protected static abstract class Builder<T extends Builder<T, U>, U extends AbstractRaftResponse> implements RaftResponse.Builder<T, U> {
+        protected Status status;
+        protected RaftError error;
 
-    @Override
-    public String toString() {
-      return toStringHelper(this)
-              .add("status", status)
-              .add("error", error)
-              .toString();
+        @Override
+        @SuppressWarnings("unchecked")
+        public T withStatus(Status status) {
+            this.status = checkNotNull(status, "status cannot be null");
+            return (T) this;
+        }
+
+        @Override
+        @SuppressWarnings("unchecked")
+        public T withError(RaftError error) {
+            this.error = checkNotNull(error, "error cannot be null");
+            return (T) this;
+        }
+
+        /**
+         * Validates the builder.
+         */
+        protected void validate() {
+            checkNotNull(status, "status cannot be null");
+            if (status == Status.ERROR) {
+                checkNotNull(error, "error cannot be null");
+            }
+        }
+
+        @Override
+        public String toString() {
+            return toStringHelper(this)
+                    .add("status", status)
+                    .add("error", error)
+                    .toString();
+        }
     }
-  }
 }

core/src/main/java/io/atomix/protocols/raft/protocol/CloseSessionRequest.java

@@ -35,11 +35,6 @@ public CloseSessionRequest(long session) {
     super(session);
   }
 
-  @Override
-  public Type type() {
-    return Type.CLOSE_SESSION;
-  }
-
   /**
    * Unregister request builder.
    */

core/src/main/java/io/atomix/protocols/raft/protocol/CommandRequest.java

@@ -53,11 +53,6 @@ public CommandRequest(long session, long sequence, byte[] bytes) {
     super(session, sequence, bytes);
   }
 
-  @Override
-  public Type type() {
-    return Type.COMMAND;
-  }
-
   @Override
   public int hashCode() {
     return Objects.hash(getClass(), session, sequence, bytes);

core/src/main/java/io/atomix/protocols/raft/protocol/ConnectRequest.java

@@ -49,11 +49,6 @@ public ConnectRequest(long session, long connection) {
     this.connection = connection;
   }
 
-  @Override
-  public Type type() {
-    return Type.CONNECT;
-  }
-
   /**
    * Returns the connecting session ID.
    *