Allow the FSM to return a response to Apply

fsm.go

@@ -8,7 +8,7 @@ import (
 // clients to make use of the replicated log
 type FSM interface {
 	// Apply log is invoked once a log entry is commited
-	Apply([]byte)
+	Apply([]byte) interface{}
 
 	// Snapshot is used to support log compaction. This call should
 	// return an FSMSnapshot which can be used to save a point-in-time

future.go

@@ -5,11 +5,17 @@ import (
 	"time"
 )
 
-// Future is used to represent an application that may occur in the future
+// Future is used to represent an action that may occur in the future
 type Future interface {
 	Error() error
 }
 
+// ApplyFuture is used for Apply() and can returns the FSM response
+type ApplyFuture interface {
+	Future
+	Response() interface{}
+}
+
 // errorFuture is used to return a static error
 type errorFuture struct {
 	err error
@@ -19,6 +25,10 @@ func (e errorFuture) Error() error {
 	return e.err
 }
 
+func (e errorFuture) Response() interface{} {
+	return nil
+}
+
 // deferError can be embedded to allow a future
 // to provide an error in the future
 type deferError struct {
@@ -53,8 +63,13 @@ func (d *deferError) respond(err error) {
 // the log is considered committed
 type logFuture struct {
 	deferError
-	log    Log
-	policy quorumPolicy
+	log      Log
+	policy   quorumPolicy
+	response interface{}
+}
+
+func (l *logFuture) Response() interface{} {
+	return l.response
 }
 
 type shutdownFuture struct {

raft.go

@@ -168,7 +168,7 @@ func NewRaft(conf *Config, fsm FSM, logs LogStore, stable StableStore, snaps Sna
 // An optional timeout can be provided to limit the amount of time we wait
 // for the command to be started. This must be run on the leader or it
 // will fail.
-func (r *Raft) Apply(cmd []byte, timeout time.Duration) Future {
+func (r *Raft) Apply(cmd []byte, timeout time.Duration) ApplyFuture {
 	var timer <-chan time.Time
 	if timeout > 0 {
 		timer = time.After(timeout)
@@ -319,14 +319,15 @@ func (r *Raft) runFSM() {
 
 		case commitTuple := <-r.fsmCommitCh:
 			// Apply the log
-			r.fsm.Apply(commitTuple.log.Data)
+			resp := r.fsm.Apply(commitTuple.log.Data)
 
 			// Update the indexes
 			lastIndex = commitTuple.log.Index
 			lastTerm = commitTuple.log.Term
 
 			// Invoke the future if given
 			if commitTuple.future != nil {
+				commitTuple.future.response = resp
 				commitTuple.future.respond(nil)
 			}
 		case <-r.shutdownCh:
@@ -944,6 +945,7 @@ func (r *Raft) installSnapshot(rpc RPC, req *InstallSnapshotRequest) {
 	case r.fsmRestoreCh <- future:
 	case <-r.shutdownCh:
 		future.respond(RaftShutdown)
+		return
 	}
 
 	// Wait for the restore to happen
@@ -968,6 +970,7 @@ func (r *Raft) installSnapshot(rpc RPC, req *InstallSnapshotRequest) {
 		log.Printf("[ERR] Failed to compact logs: %v", err)
 	}
 
+	log.Printf("[INFO] Installed remote snapshot")
 	resp.Success = true
 	return
 }

raft_test.go

@@ -326,6 +326,11 @@ func TestRaft_SingleNode(t *testing.T) {
 		t.Fatalf("err: %v", err)
 	}
 
+	// Check the response
+	if future.Response().(int) != 1 {
+		t.Fatalf("bad response: %v", future.Response())
+	}
+
 	// Check that it is applied to the FSM
 	if len(c.fsms[0].logs) != 1 {
 		t.Fatalf("did not apply to FSM!")