Simple key-value store

barge-jax-rs/pom.xml

@@ -126,40 +126,4 @@
 
   </dependencies>
 
-  <build>
-    <finalName>barge-http</finalName>
-    <plugins>
-      <plugin>
-        <artifactId>maven-shade-plugin</artifactId>
-        <version>2.1</version>
-        <executions>
-          <execution>
-            <phase>package</phase>
-            <goals>
-              <goal>shade</goal>
-            </goals>
-            <configuration>
-              <filters>
-                <filter>
-                  <artifact>*:*</artifact>
-                  <excludes>
-                    <exclude>META-INF/*.SF</exclude>
-                    <exclude>META-INF/*.DSA</exclude>
-                    <exclude>META-INF/*.RSA</exclude>
-                  </excludes>
-                </filter>
-              </filters>
-              <transformers>
-                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
-                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
-                  <mainClass>org.robotninjas.barge.jaxrs.ws.RaftJettyServer</mainClass>
-                </transformer>
-              </transformers>
-            </configuration>
-          </execution>
-        </executions>
-      </plugin>
-    </plugins>
-  </build>
-
 </project>

barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/Leaders.java

@@ -0,0 +1,69 @@
+package org.robotninjas.barge.jaxrs;
+
+import org.robotninjas.barge.state.Raft;
+import org.robotninjas.barge.utils.Prober;
+
+import java.net.URI;
+
+import java.util.concurrent.Callable;
+
+import javax.ws.rs.client.Client;
+
+
+/**
+ * Utility to query for leader in a cluster and wait for cluster to reach a stable state with one leader.
+ */
+public class Leaders {
+
+  private final URI[] uris;
+
+  public Leaders(URI[] uris) {
+    this.uris = uris;
+  }
+
+  public boolean isLeader(Client client, URI uri) {
+    return client.target(uri).path("/raft/state").request().get(Raft.StateType.class).equals(Raft.StateType.LEADER);
+  }
+
+  /**
+   * Retrieves the URI of the current leader in the cluster.
+   *
+   * @param client HTTP client to use
+   * @return the URI of the current leader.
+   * @throws  java.lang.IllegalStateException  if there is no current leader in the cluster.
+   */
+  public URI getLeader(Client client) {
+
+    if (isLeader(client, uris[0]))
+      return uris[0];
+
+    if (isLeader(client, uris[1]))
+      return uris[1];
+
+    if (isLeader(client, uris[2]))
+      return uris[2];
+
+    throw new IllegalStateException("expected one server to be a leader");
+  }
+
+  /**
+   * Wait a certain amount of time for a leader to emerge in this cluster.
+   * @param client the HTTP client to use.
+   * @param timeoutInMs timeout in milliseconds before giving up.
+   */
+  public void waitForALeader(final Client client, int timeoutInMs) {
+    new Prober(new Callable<Boolean>() {
+        @Override
+        public Boolean call() throws Exception {
+
+          for (URI uri : uris) {
+
+            if (isLeader(client, uri))
+              return true;
+          }
+
+          return false;
+        }
+      }).probe(timeoutInMs);
+  }
+}

barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftApplication.java

@@ -3,6 +3,7 @@
 import com.google.common.base.Function;
 import com.google.common.base.Optional;
 import com.google.common.collect.Lists;
+import com.google.common.collect.Sets;
 import com.google.inject.Guice;
 import com.google.inject.Injector;
 import org.glassfish.hk2.api.Factory;
@@ -24,8 +25,9 @@
 import java.io.IOException;
 import java.net.URI;
 import java.nio.ByteBuffer;
-import java.util.Collections;
+import java.util.Arrays;
 import java.util.List;
+import java.util.Set;
 
 
 /**
@@ -37,40 +39,33 @@ public class RaftApplication {
   private final int serverIndex;
   private final URI[] uris;
   private final File logDir;
-
   private final List<StateTransitionListener> transitionListeners;
   private final List<RaftProtocolListener> protocolListeners;
+  private final Set<Class<?>> resources = Sets.newHashSet();
+  private final Set<Object> instances = Sets.newHashSet();
+  private final StateMachine stateMachine;
 
   private Optional<Injector> injector = Optional.absent();
 
-  public RaftApplication(int serverIndex, URI[] uris, File logDir, Iterable<StateTransitionListener> transitionListener, Iterable<RaftProtocolListener> protocolListener) {
+  public RaftApplication(int serverIndex, URI[] uris, File logDir, StateMachine stateMachine, Set<Class<?>> resources,
+      Set<Object> instances,
+      Iterable<StateTransitionListener> transitionListener, Iterable<RaftProtocolListener> protocolListener) {
     this.serverIndex = serverIndex;
     this.uris = uris;
     this.logDir = logDir;
+    this.stateMachine = stateMachine;
+    this.resources.addAll(resources);
+    this.instances.addAll(instances);
     this.transitionListeners = Lists.newArrayList(transitionListener);
     this.protocolListeners = Lists.newArrayList(protocolListener);
   }
 
-  public RaftApplication(int serverIndex, URI[] uris, File logDir) {
-    this(serverIndex,uris,logDir, Collections.<StateTransitionListener>emptyList(),Collections.<RaftProtocolListener>emptyList());
-  }
-
   public ResourceConfig makeResourceConfig() {
-    ClusterConfig clusterConfig = HttpClusterConfig.from(new HttpReplica(uris[serverIndex]),
-      remotes());
+    ClusterConfig clusterConfig = HttpClusterConfig.from(new HttpReplica(uris[serverIndex]), remotes());
 
     if (!logDir.exists() && !logDir.mkdirs())
       logger.warn("failed to create directories for storing logs, bad things will happen");
 
-    StateMachine stateMachine = new StateMachine() {
-      int i = 0;
-
-      @Override
-      public Object applyOperation(@Nonnull ByteBuffer entry) {
-        return i++;
-      }
-    };
-
     final JaxRsRaftModule raftModule = new JaxRsRaftModule(clusterConfig, logDir, stateMachine, 1500, transitionListeners, protocolListeners);
 
     injector = Optional.of(Guice.createInjector(raftModule));
@@ -104,6 +99,8 @@ public void dispose(Raft raft) {
 
     resourceConfig.register(BargeResource.class);
     resourceConfig.register(Jackson.customJacksonProvider());
+    resourceConfig.registerClasses(resources);
+    resourceConfig.registerInstances(instances);
     resourceConfig.register(binder);
 
     return resourceConfig;
@@ -125,16 +122,82 @@ public void clean() throws IOException {
 
   public void stop() {
     injector.transform(new Function<Injector, Object>() {
-      @Nullable
-      @Override
-      public Object apply(@Nullable Injector input) {
-        Raft instance = null;
-        if (input != null) {
-          instance = input.getInstance(Raft.class);
-          instance.stop();
+        @Nullable
+        @Override
+        public Object apply(@Nullable Injector input) {
+          Raft instance = null;
+
+          if (input != null) {
+            instance = input.getInstance(Raft.class);
+            instance.stop();
+          }
+
+          return instance;
         }
-        return instance;
+      });
+  }
+
+  public static class Builder {
+    private int serverIndex;
+    private URI[] uris;
+    private File logDir;
+    private StateTransitionListener[] transitionListeners = new StateTransitionListener[0];
+    private Iterable<RaftProtocolListener> protocolListeners = Lists.newArrayList();
+    private Set<Class<?>> resources = Sets.newHashSet();
+    private Set<Object> instances = Sets.newHashSet();
+    private StateMachine stateMachine = new StateMachine() {
+      int i = 0;
+
+      @Override
+      public Object applyOperation(@Nonnull ByteBuffer entry) {
+        return i++;
       }
-    });
+    };
+
+    public Builder setServerIndex(int serverIndex) {
+      this.serverIndex = serverIndex;
+
+      return this;
+    }
+
+    public Builder setUris(URI[] uris) {
+      this.uris = uris;
+
+      return this;
+    }
+
+    public Builder setLogDir(File logDir) {
+      this.logDir = logDir;
+
+      return this;
+    }
+
+    public Builder setTransitionListeners(StateTransitionListener... listeners) {
+      this.transitionListeners = listeners;
+
+      return this;
+    }
+
+    public RaftApplication build() {
+      return new RaftApplication(serverIndex, uris, logDir, stateMachine, resources, instances, Arrays.asList(transitionListeners), protocolListeners);
+    }
+
+    public Builder register(Class<?> resource) {
+      this.resources.add(resource);
+
+      return this;
+    }
+
+    public Builder setStateMachine(StateMachine stateMachine) {
+      this.stateMachine = stateMachine;
+
+      return this;
+    }
+
+    public Builder registerInstance(Object o) {
+      this.instances.add(o);
+
+      return this;
+    }
   }
 }

barge-jax-rs/src/main/java/org/robotninjas/barge/jaxrs/RaftJdkServer.java

@@ -16,14 +16,18 @@
 package org.robotninjas.barge.jaxrs;
 
 import com.google.common.base.Throwables;
+
 import com.sun.net.httpserver.HttpServer;
+
 import org.glassfish.jersey.jdkhttp.JdkHttpServerFactory;
 
-import javax.ws.rs.core.UriBuilder;
 import java.io.File;
 import java.io.IOException;
+
 import java.net.URI;
 
+import javax.ws.rs.core.UriBuilder;
+
 
 /**
  * A dedicated server for an instance of Raft using JDK's embedded HTTP server.
@@ -39,7 +43,7 @@ public class RaftJdkServer implements RaftServer<RaftJdkServer> {
   public RaftJdkServer(int serverIndex, URI[] uris, File logDir) {
     this.serverIndex = serverIndex;
     this.uris = uris;
-    this.application = new RaftApplication(serverIndex, uris, logDir);
+    this.application = new RaftApplication.Builder().setServerIndex(serverIndex).setUris(uris).setLogDir(logDir).build();
   }