test2

src/main/java/com/yuandian/dataflow/Server.java

@@ -1,13 +1,22 @@
 package com.yuandian.dataflow;
 
+import com.yuandian.dataflow.rpc.DataFlowGrpc;
+import com.yuandian.dataflow.rpc.DataFlowGrpc.DataFlowImplBase;
+
 /**
  * Hello world!
  *
  */
 public class Server 
 {
+    public class ServerImpl extends DataFlowImplBase {
+        
+    }
+
+
     public static void main( String[] args )
     {
+        
         System.out.println( "Hello World!" );
-    }
+    }                    
 }

src/main/java/com/yuandian/dataflow/statemachine/StateMachine.java

@@ -0,0 +1,99 @@
+package com.yuandian.dataflow.statemachine;
+
+ 
+
+import java.io.File;
+import java.io.IOException;
+import java.nio.ByteBuffer;
+import java.util.concurrent.atomic.AtomicLong;
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+import com.alipay.remoting.exception.CodecException;
+import com.alipay.remoting.serialization.SerializerManager;
+import com.alipay.sofa.jraft.Closure;
+import com.alipay.sofa.jraft.Iterator;
+import com.alipay.sofa.jraft.Status;
+import com.alipay.sofa.jraft.core.StateMachineAdapter;
+import com.alipay.sofa.jraft.error.RaftError;
+import com.alipay.sofa.jraft.error.RaftException;
+import com.alipay.sofa.jraft.storage.snapshot.SnapshotReader;
+import com.alipay.sofa.jraft.storage.snapshot.SnapshotWriter;
+import com.alipay.sofa.jraft.util.Utils;
+
+/**
+ * Counter state machine.
+ *
+ * @author boyan (boyan@alibaba-inc.com)
+ *
+ * 2018-Apr-09 4:52:31 PM
+ */
+public class StateMachine extends StateMachineAdapter {
+
+    private static final Logger LOG        = LoggerFactory.getLogger(StateMachine.class);
+
+    /**
+     * Counter value
+     */
+    private final AtomicLong    value      = new AtomicLong(0);
+    /**
+     * Leader term
+     */
+    private final AtomicLong    leaderTerm = new AtomicLong(-1);
+
+    public boolean isLeader() {
+        return this.leaderTerm.get() > 0;
+    }
+
+    /**
+     * Returns current value.
+     */
+    public long getValue() {
+        return this.value.get();
+    }
+
+    @Override
+    public void onApply(final Iterator iter) {
+        while (iter.hasNext()) {
+         
+            if (iter.done() != null) {
+                // This task is applied by this node, get value from closure to avoid additional parsing.
+                
+            } else {
+                // Have to parse FetchAddRequest from this user log.
+              
+            }
+      
+            iter.next();
+        }
+    }
+
+    @Override
+    public void onSnapshotSave(final SnapshotWriter writer, final Closure done) {
+        return;
+    }
+
+    @Override
+    public void onError(final RaftException e) {
+        LOG.error("Raft error: {}", e, e);
+    }
+
+    @Override
+    public boolean onSnapshotLoad(final SnapshotReader reader) {
+        
+        return true;
+    }
+
+    @Override
+    public void onLeaderStart(final long term) {
+        this.leaderTerm.set(term);
+        super.onLeaderStart(term);
+
+    }
+
+    @Override
+    public void onLeaderStop(final Status status) {
+        this.leaderTerm.set(-1);
+        super.onLeaderStop(status);
+    }
+
+}