Make sessions remote only.

cluster/src/main/java/net/kuujo/copycat/cluster/ManagedCluster.java

@@ -157,6 +157,9 @@ public CompletableFuture<Void> configure(MemberInfo... membersInfo) {
    * Registers a member.
    */
   public CompletableFuture<Member> register(Session session) {
+    if (session.member().id() == localMember.id())
+      return CompletableFuture.completedFuture(localMember);
+
     ManagedRemoteMember member = remoteMembers.containsKey(session.member().id()) ? remoteMembers.get(session.member().id()) : createMember(session.member());
     return member.connect().thenApply(v -> {
       member.status = Member.Status.ALIVE;
@@ -254,7 +257,7 @@ public ManagedLocalMember member() {
   }
 
   @Override
-  public ManagedMember member(int id) {
+  public Member member(int id) {
     if (localMember.id() == id)
       return localMember;
     ManagedMember member = remoteMembers.get(id);

cluster/src/main/java/net/kuujo/copycat/cluster/ManagedMember.java

@@ -28,7 +28,6 @@ public abstract class ManagedMember implements Member {
   protected final MemberInfo info;
   protected Type type;
   protected Status status = Status.DEAD;
-  protected Session session;
   protected final ExecutionContext context;
 
   protected ManagedMember(MemberInfo info, Type type, ExecutionContext context) {
@@ -65,11 +64,6 @@ public MemberInfo info() {
     return info;
   }
 
-  @Override
-  public Session session() {
-    return session;
-  }
-
   @Override
   public boolean equals(Object object) {
     return object instanceof Member && ((Member) object).id() == info.id();

cluster/src/main/java/net/kuujo/copycat/cluster/ManagedRemoteMember.java

@@ -25,11 +25,17 @@
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
 public abstract class ManagedRemoteMember extends ManagedMember implements RemoteMember {
+  protected Session session;
 
   protected ManagedRemoteMember(MemberInfo info, Type type, ExecutionContext context) {
     super(info, type, context);
   }
 
+  @Override
+  public Session session() {
+    return session;
+  }
+
   /**
    * Connects the remote client.
    *

cluster/src/main/java/net/kuujo/copycat/cluster/Member.java

@@ -93,13 +93,6 @@ static enum Status {
    */
   MemberInfo info();
 
-  /**
-   * Returns the member session.
-   *
-   * @return The member session.
-   */
-  Session session();
-
   /**
    * Sends a message to the member.<p>
    *

cluster/src/main/java/net/kuujo/copycat/cluster/RemoteMember.java

@@ -21,4 +21,12 @@
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
 public interface RemoteMember extends Member {
+
+  /**
+   * Returns the member session.
+   *
+   * @return The member session.
+   */
+  Session session();
+
 }

raft/src/main/java/net/kuujo/copycat/raft/state/RaftContext.java

@@ -56,6 +56,7 @@ public class RaftContext implements Managed<RaftContext> {
   private CompletableFuture<RaftContext> openFuture;
   private long electionTimeout = 500;
   private long heartbeatInterval = 250;
+  private long session;
   private int leader;
   private long term;
   private int lastVotedFor;
@@ -194,6 +195,26 @@ public RaftContext removeElectionListener(EventListener<Member> listener) {
     return this;
   }
 
+  /**
+   * Sets the state session.
+   *
+   * @param session The state session.
+   * @return The Raft context.
+   */
+  RaftContext setSession(long session) {
+    this.session = session;
+    return this;
+  }
+
+  /**
+   * Returns the state session.
+   *
+   * @return The state session.
+   */
+  public long getSession() {
+    return session;
+  }
+
   /**
    * Sets the state leader.
    *

raft/src/main/java/net/kuujo/copycat/raft/state/RemoteState.java

@@ -86,6 +86,7 @@ private CompletableFuture<Long> register(List<Member> members, CompletableFuture
       if (error == null) {
         context.setTerm(response.term());
         context.setLeader(response.leader());
+        context.setSession(response.session());
         context.getCluster().configure(response.members().toArray(new TypedMemberInfo[response.members().size()])).whenComplete((configureResult, configureError) -> {
           if (configureError == null) {
             future.complete(response.session());
@@ -139,7 +140,7 @@ private CompletableFuture<Void> keepAlive(List<Member> members, CompletableFutur
     }
 
     KeepAliveRequest request = KeepAliveRequest.builder()
-      .withSession(context.getCluster().member().session().id())
+      .withSession(context.getSession())
       .build();
     member.<KeepAliveRequest, KeepAliveResponse>send(context.getTopic(), request).whenComplete((response, error) -> {
       if (error == null) {

test/src/main/java/net/kuujo/copycat/cluster/TestCluster.java

@@ -128,7 +128,7 @@ public TestCluster build() {
         info = new TestMember.Info(memberId, address);
       }
 
-      TestLocalMember localMember = new TestLocalMember(info, type, serializer != null ? serializer : new Serializer(), new ExecutionContext(String.format("copycat-cluster-%d", memberId)));
+      TestLocalMember localMember = new TestLocalMember(info, type, member != null, serializer != null ? serializer : new Serializer(), new ExecutionContext(String.format("copycat-cluster-%d", memberId)));
       return new TestCluster(localMember, members.values().stream().map(m -> (TestRemoteMember) m).collect(Collectors.toList()), registry, serializer);
     }
   }

test/src/main/java/net/kuujo/copycat/cluster/TestLocalMember.java

@@ -35,8 +35,8 @@ public class TestLocalMember extends ManagedLocalMember implements TestMember {
   private final Map<String, HandlerHolder> handlers = new HashMap<>();
   private TestMemberRegistry registry;
 
-  TestLocalMember(TestMember.Info info, Member.Type type, Serializer serializer, ExecutionContext context) {
-    super(info, type, context);
+  TestLocalMember(TestMember.Info info, Member.Type type, boolean seed, Serializer serializer, ExecutionContext context) {
+    super(info, type, seed, context);
     this.serializer = serializer;
     this.info = info;
   }