WIP: Incorporate snapshot sending/receiving into RaftAlgorithm

libraft-agent/src/main/java/io/libraft/agent/RaftAgent.java

@@ -50,6 +50,7 @@
 import io.libraft.agent.protocol.RaftRPC;
 import io.libraft.agent.rpc.RaftNetworkClient;
 import io.libraft.algorithm.RaftAlgorithm;
+import io.libraft.algorithm.RaftConstants;
 import io.libraft.algorithm.StorageException;
 import org.apache.tomcat.jdbc.pool.DataSource;
 import org.jboss.netty.channel.socket.ClientSocketChannelFactory;
@@ -251,6 +252,7 @@ private RaftAgent(RaftConfiguration configuration, RaftListener raftListener) {
                 getMemberIds(cluster),
                 snapshotsConfiguration.getMinEntriesToSnapshot(),
                 snapshotsConfiguration.getSnapshotCheckInterval(),
+                RaftConstants.MAX_LOG_ENTRIES_PER_APPEND_ENTRIES, // do not want to expose this right now
                 configuration.getRPCTimeout(),
                 configuration.getMinElectionTimeout(),
                 configuration.getAdditionalElectionTimeoutRange(),

libraft-agent/src/main/java/io/libraft/agent/rpc/RaftNetworkClient.java

@@ -58,6 +58,7 @@
 import org.slf4j.LoggerFactory;
 
 import javax.annotation.Nullable;
+import java.io.InputStream;
 import java.net.InetSocketAddress;
 import java.net.SocketAddress;
 import java.util.Collection;
@@ -464,4 +465,14 @@ public void appendEntries(String server, long term, long commitIndex, long prevL
     public void appendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) throws RPCException {
         write(server, new AppendEntriesReply(self.getId(), server, term, prevLogIndex, entryCount, applied));
     }
+
+    @Override
+    public void snapshotChunk(String server, long term, long snapshotTerm, long snapshotIndex, int seqnum, InputStream chunkInputStream) throws RPCException {
+        // FIXME (AG): noop
+    }
+
+    @Override
+    public void snapshotChunkReply(String server, long term, long snapshotTerm, long snapshotIndex, int nextExpectedSeqnum) throws RPCException {
+        // FIXME (AG): noop
+    }
 }

libraft-core/src/main/java/io/libraft/algorithm/ClusterCommittedCommand.java

@@ -0,0 +1,94 @@
+/*
+ * Copyright (c) 2013 - 2014, Allen A. George <allen dot george at gmail dot com>
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in the
+ *       documentation and/or other materials provided with the distribution.
+ *     * Neither the name of libraft nor the
+ *       names of its contributors may be used to endorse or promote products
+ *       derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.libraft.algorithm;
+
+import com.google.common.base.Objects;
+import io.libraft.Command;
+import io.libraft.CommittedCommand;
+
+import javax.annotation.Nullable;
+
+// Represents a command that has been committed to the Raft cluster.
+final class ClusterCommittedCommand implements CommittedCommand {
+
+    private final long index;
+    private final Command command;
+
+    /**
+     * Constructor.
+     *
+     * @param index index > 0 in the Raft log of the committed {@code Command}
+     * @param command the committed {@code Command} instance
+     *
+     * @see Command
+     */
+    ClusterCommittedCommand(long index, Command command) {
+        this.index = index;
+        this.command = command;
+    }
+
+    @Override
+    public Type getType() {
+        return Type.COMMAND;
+    }
+
+    @Override
+    public long getIndex() {
+        return index;
+    }
+
+    @Override
+    public Command getCommand() {
+        return command;
+    }
+
+    @Override
+    public boolean equals(@Nullable Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        ClusterCommittedCommand other = (ClusterCommittedCommand) o;
+
+        return index == other.index && command.equals(other.command);
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(index, command);
+    }
+
+    @Override
+    public String toString() {
+        return Objects
+                .toStringHelper(this)
+                .add("index", index)
+                .add("command", command)
+                .toString();
+    }
+}

libraft-core/src/main/java/io/libraft/algorithm/ClusterCommittedNoop.java

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) 2013 - 2014, Allen A. George <allen dot george at gmail dot com>
+ *
+ * All rights reserved.
+ *
+ * Redistribution and use in source and binary forms, with or without
+ * modification, are permitted provided that the following conditions are met:
+ *     * Redistributions of source code must retain the above copyright
+ *       notice, this list of conditions and the following disclaimer.
+ *     * Redistributions in binary form must reproduce the above copyright
+ *       notice, this list of conditions and the following disclaimer in the
+ *       documentation and/or other materials provided with the distribution.
+ *     * Neither the name of libraft nor the
+ *       names of its contributors may be used to endorse or promote products
+ *       derived from this software without specific prior written permission.
+ *
+ * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
+ * ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
+ * WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
+ * DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE FOR
+ * ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES
+ * (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES;
+ * LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND
+ * ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT
+ * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE OF THIS
+ * SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
+ */
+
+package io.libraft.algorithm;
+
+import com.google.common.base.Objects;
+import io.libraft.Committed;
+
+import javax.annotation.Nullable;
+
+// Represents a noop that has been committed to the Raft cluster.
+final class ClusterCommittedNoop implements Committed {
+
+    private final long index;
+
+    /**
+     * Constructor.
+     *
+     * @param index index > 0 in the Raft log of the committed noop
+     */
+    ClusterCommittedNoop(long index) {
+        this.index = index;
+    }
+
+    @Override
+    public Type getType() {
+        return Type.SKIP;
+    }
+
+    @Override
+    public long getIndex() {
+        return index;
+    }
+
+    @Override
+    public boolean equals(@Nullable Object o) {
+        if (this == o) return true;
+        if (o == null || getClass() != o.getClass()) return false;
+
+        ClusterCommittedNoop other = (ClusterCommittedNoop) o;
+        return index == other.index;
+    }
+
+    @Override
+    public int hashCode() {
+        return Objects.hashCode(index);
+    }
+
+    @Override
+    public String toString() {
+        return Objects
+                .toStringHelper(this)
+                .add("index", index)
+                .toString();
+    }
+}

libraft-core/src/main/java/io/libraft/algorithm/RPCReceiver.java

@@ -29,6 +29,7 @@
 package io.libraft.algorithm;
 
 import javax.annotation.Nullable;
+import java.io.InputStream;
 import java.util.Collection;
 
 /**
@@ -55,6 +56,8 @@
  */
 public interface RPCReceiver {
 
+    // FIXME (AG): we will avoid wasted work in RaftAlgorithm if we have some notion of whether server is 'available' or not
+
     /**
      * Indicates that a RequestVote was received.
      *
@@ -108,4 +111,8 @@ public interface RPCReceiver {
      *                to its log. false otherwise
      */
     void onAppendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied);
+
+    void onSnapshotChunk(String server, long term, long snapshotTerm, long snapshotIndex, int seqnum, @Nullable InputStream chunkInputStream);
+
+    void onSnapshotChunkReply(String server, long term, long snapshotTerm, long snapshotIndex, int nextExpectedSeqnum);
 }

libraft-core/src/main/java/io/libraft/algorithm/RPCSender.java

@@ -29,6 +29,7 @@
 package io.libraft.algorithm;
 
 import javax.annotation.Nullable;
+import java.io.InputStream;
 import java.util.Collection;
 
 /**
@@ -110,4 +111,8 @@ public interface RPCSender {
      * @see LogEntry
      */
     void appendEntriesReply(String server, long term, long prevLogIndex, long entryCount, boolean applied) throws RPCException;
+
+    void snapshotChunk(String server, long term, long snapshotTerm, long snapshotIndex, int seqnum, @Nullable InputStream chunkInputStream) throws RPCException;
+
+    void snapshotChunkReply(String server, long term, long snapshotTerm, long snapshotIndex, int nextExpectedSeqnum) throws RPCException;
 }