Refactor context to support pluggable contexts and use the Netty even…

…t loop.

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

@@ -16,7 +16,7 @@
 package net.kuujo.copycat.cluster;
 
 import net.kuujo.alleycat.Alleycat;
-import net.kuujo.copycat.util.ExecutionContext;
+import net.kuujo.copycat.util.Context;
 import net.kuujo.copycat.util.Managed;
 
 import java.util.Random;
@@ -30,8 +30,8 @@ public abstract class ManagedMember implements Member, Managed<Member> {
   protected final MemberInfo info;
   protected Type type;
   protected Status status = Status.DEAD;
-  protected ExecutionContext context;
-  protected Alleycat alleycat;
+  protected Context context;
+  protected Alleycat serializer;
 
   protected ManagedMember(MemberInfo info, Type type) {
     this.info = info;
@@ -41,16 +41,16 @@ protected ManagedMember(MemberInfo info, Type type) {
   /**
    * Sets the member context.
    */
-  void setContext(ExecutionContext context) {
+  void setContext(Context context) {
     this.context = context;
-    this.alleycat = context.alleycat();
+    this.serializer = context.serializer();
   }
 
   /**
    * Returns the current execution context.
    */
-  protected ExecutionContext getContext() {
-    ExecutionContext context = ExecutionContext.currentContext();
+  protected Context getContext() {
+    Context context = Context.currentContext();
     return context != null ? context : this.context;
   }
 

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

@@ -16,7 +16,7 @@
 package net.kuujo.copycat.cluster;
 
 import net.kuujo.alleycat.Alleycat;
-import net.kuujo.copycat.util.ExecutionContext;
+import net.kuujo.copycat.util.Context;
 import net.kuujo.copycat.util.Managed;
 
 import java.util.*;
@@ -43,7 +43,7 @@ public abstract class ManagedMembers implements Members, Managed<Members> {
 
   protected ManagedMembers(Collection<? extends ManagedMember> remoteMembers, Alleycat alleycat) {
     remoteMembers.forEach(m -> {
-      ((ManagedMember)m).setContext(new ExecutionContext("copycat-cluster-" + m.id(), alleycat));
+      ((ManagedMember)m).setContext(Context.createContext("copycat-cluster-%d", alleycat));
       this.members.put(m.id(), m);
       this.sortedMembers.add(m);
     });

cluster/src/main/java/net/kuujo/copycat/util/Context.java

@@ -0,0 +1,118 @@
+/*
+ * Copyright 2015 the original author or authors.
+ *
+ * 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 net.kuujo.copycat.util;
+
+import net.kuujo.alleycat.Alleycat;
+
+import java.util.ServiceLoader;
+import java.util.concurrent.Executor;
+import java.util.concurrent.ScheduledFuture;
+import java.util.concurrent.TimeUnit;
+
+/**
+ * Execution context.
+ *
+ * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
+ */
+public abstract class Context implements Executor, AutoCloseable {
+  private static final ContextFactory factory = ServiceLoader.load(ContextFactory.class).iterator().next();
+
+  /**
+   * Creates a new context.
+   *
+   * @param name The context name.
+   * @param serializer The Alleycat serializer.
+   * @return A new context.
+   */
+  public static Context createContext(String name, Alleycat serializer) {
+    return factory.createContext(name, serializer);
+  }
+
+  /**
+   * Returns the current context.
+   *
+   * @return The current context.
+   */
+  public static Context currentContext() {
+    Thread thread = Thread.currentThread();
+    return thread instanceof CopycatThread ? ((CopycatThread) thread).getContext() : null;
+  }
+
+  private final String name;
+  private final Alleycat serializer;
+
+  protected Context(String name, CopycatThread thread, Alleycat serializer) {
+    if (thread == null)
+      throw new NullPointerException("thread cannot be null");
+    thread.setName(name);
+    thread.setContext(this);
+    this.name = name;
+    this.serializer = serializer;
+  }
+
+  /**
+   * Checks that the current thread is the correct context thread.
+   */
+  public void checkThread() {
+    Thread thread = Thread.currentThread();
+    if (!(thread instanceof CopycatThread && ((CopycatThread) thread).getContext() == this)) {
+      throw new IllegalStateException("not running on the correct thread");
+    }
+  }
+
+  /**
+   * Returns the context name.
+   *
+   * @return The context name.
+   */
+  public String name() {
+    return name;
+  }
+
+  /**
+   * Returns the context serializer.
+   *
+   * @return The context serializer.
+   */
+  public Alleycat serializer() {
+    return serializer;
+  }
+
+  /**
+   * Schedules a runnable on the context.
+   *
+   * @param runnable The runnable to schedule.
+   * @param delay The delay at which to schedule the runnable.
+   * @param unit The time unit.
+   */
+  public abstract ScheduledFuture<?> schedule(Runnable runnable, long delay, TimeUnit unit);
+
+  /**
+   * Schedules a runnable at a fixed rate on the context.
+   *
+   * @param runnable The runnable to schedule.
+   * @param delay The delay at which to schedule the runnable.
+   * @param unit The time unit.
+   */
+  public abstract ScheduledFuture<?> scheduleAtFixedRate(Runnable runnable, long delay, long rate, TimeUnit unit);
+
+  /**
+   * Closes the context.
+   */
+  @Override
+  public abstract void close();
+
+}

cluster/src/main/java/net/kuujo/copycat/util/ThreadChecker.java → cluster/src/main/java/net/kuujo/copycat/util/ContextFactory.java

@@ -15,26 +15,22 @@
  */
 package net.kuujo.copycat.util;
 
+import net.kuujo.alleycat.Alleycat;
+
 /**
- * Context thread checker.
+ * Context factory.
  *
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
-public class ThreadChecker {
-  private final ExecutionContext context;
-
-  public ThreadChecker(ExecutionContext context) {
-    this.context = context;
-  }
+public interface ContextFactory {
 
   /**
-   * Checks that the current thread is the correct context thread.
+   * Creates a new execution context.
+   *
+   * @param name The context name.
+   * @param serializer The Alleycat serializer.
+   * @return The execution context.
    */
-  public void checkThread() {
-    Thread thread = Thread.currentThread();
-    if (!(thread instanceof net.kuujo.copycat.util.CopycatThread && ((net.kuujo.copycat.util.CopycatThread) thread).getContext() == context)) {
-      throw new IllegalStateException("not running on the correct thread");
-    }
-  }
+  Context createContext(String name, Alleycat serializer);
 
 }

cluster/src/main/java/net/kuujo/copycat/util/CopycatThread.java

@@ -23,7 +23,7 @@
  * @author <a href="http://github.com/kuujo">Jordan Halterman</a>
  */
 public class CopycatThread extends Thread {
-  private WeakReference<ExecutionContext> context;
+  private WeakReference<Context> context;
 
   public CopycatThread(Runnable target, String name) {
     super(target, name);
@@ -32,14 +32,14 @@ public CopycatThread(Runnable target, String name) {
   /**
    * Sets the thread context.
    */
-  public void setContext(ExecutionContext context) {
+  public void setContext(Context context) {
     this.context = new WeakReference<>(context);
   }
 
   /**
    * Returns the thread context.
    */
-  public ExecutionContext getContext() {
+  public Context getContext() {
     return context.get();
   }
 

netty/src/main/java/net/kuujo/copycat/cluster/NettyCluster.java

@@ -20,7 +20,7 @@
 import net.kuujo.alleycat.Alleycat;
 import net.kuujo.alleycat.ServiceLoaderResolver;
 import net.kuujo.copycat.ConfigurationException;
-import net.kuujo.copycat.util.ExecutionContext;
+import net.kuujo.copycat.util.Context;
 
 import java.net.InetSocketAddress;
 import java.util.Collection;
@@ -55,7 +55,7 @@ public NettyCluster(EventLoopGroup eventLoopGroup, NettyLocalMember localMember,
   protected ManagedRemoteMember createMember(MemberInfo info) {
     ManagedRemoteMember remoteMember = new NettyRemoteMember((NettyMemberInfo) info, Member.Type.PASSIVE)
       .setEventLoopGroup(eventLoopGroup);
-    remoteMember.setContext(new ExecutionContext(String.format("copycat-cluster-%d", info.id()), alleycat));
+    remoteMember.setContext(Context.createContext(String.format("copycat-cluster-%d", info.id()), alleycat));
     return remoteMember;
   }