howmuchcomputer
diff --git a/‎.gitignore‎
Lines changed: 1 addition & 0 deletions b/‎.gitignore‎
Lines changed: 1 addition & 0 deletions
diff --git a/‎consensus-logs/copycat-1-1.log‎
-1 MB b/‎consensus-logs/copycat-1-1.log‎
-1 MB
diff --git a/‎consensus-logs/copycat-1-20171225114448.snapshot‎
-128 Bytes b/‎consensus-logs/copycat-1-20171225114448.snapshot‎
-128 Bytes
diff --git a/‎consensus-logs/copycat.meta‎
-128 Bytes b/‎consensus-logs/copycat.meta‎
-128 Bytes
diff --git a/‎src/main/java/org/tron/consensus/client/Client.java‎
Lines changed: 20 additions & 3 deletions b/‎src/main/java/org/tron/consensus/client/Client.java‎
Lines changed: 20 additions & 3 deletions
diff --git a/‎src/main/java/org/tron/consensus/common/Listen.java‎
Lines changed: 7 additions & 0 deletions b/‎src/main/java/org/tron/consensus/common/Listen.java‎
Lines changed: 7 additions & 0 deletions
diff --git a/‎src/main/java/org/tron/consensus/common/MapstateMachine.java‎
Lines changed: 41 additions & 3 deletions b/‎src/main/java/org/tron/consensus/common/MapstateMachine.java‎
Lines changed: 41 additions & 3 deletions
@@ -13,6 +13,7 @@ gradlew.bat
 
 # log
 logs
+consensus-logs
 
 # lombok
 out
 
@@ -4,14 +4,17 @@
 import io.atomix.catalyst.transport.netty.NettyTransport;
 import io.atomix.copycat.client.ConnectionStrategies;
 import io.atomix.copycat.client.CopycatClient;
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.tron.consensus.common.GetQuery;
 import org.tron.consensus.common.PutCommand;
+import org.tron.overlay.kafka.ConsumerWorker;
 
 import java.util.Arrays;
 import java.util.Collection;
 import java.util.concurrent.CompletableFuture;
 
-public class Client {
+public class Client{
 
     private static CopycatClient client = null;
 
@@ -47,10 +50,24 @@ public static void putMessage(String[] args) {
     }
 
     public static void getMessage1(String key) {
+
         client.submit(new GetQuery(key)).thenAccept(result -> {
-            System.out.println("Consensus " + key + " is: " +
-                    result);
+            System.out.println("Consensus " + key + " is: " + result);
         });
+//        Thread thread = new Thread(() -> {
+//            while (true) {
+//                try {
+//                    client.submit(new GetQuery(key)).thenAccept(result -> {
+//                        System.out.println("Consensus " + key + " is: " +
+//                                result);
+//                    });
+//                    Thread.sleep(5000);
+//                } catch (InterruptedException e) {
+//                    e.printStackTrace();
+//                }
+//            }
+//        });
+        //thread.start();
     }
     public static void getMessage(String key) {
         Object result = client.submit(new GetQuery(key)).join();
 
@@ -0,0 +1,7 @@
+package org.tron.consensus.common;
+
+import io.atomix.copycat.Operation;
+
+public class Listen implements Operation {
+
+}
@@ -1,27 +1,40 @@
 package org.tron.consensus.common;
 
+import io.atomix.catalyst.concurrent.Scheduled;
 import io.atomix.copycat.server.Commit;
 import io.atomix.copycat.server.Snapshottable;
 import io.atomix.copycat.server.StateMachine;
+import io.atomix.copycat.server.session.ServerSession;
+import io.atomix.copycat.server.session.SessionListener;
 import io.atomix.copycat.server.storage.snapshot.SnapshotReader;
 import io.atomix.copycat.server.storage.snapshot.SnapshotWriter;
 
 import java.util.HashMap;
 import java.util.Map;
+import java.util.Set;
+import java.util.HashSet;
 
-public class MapstateMachine extends StateMachine implements Snapshottable {
+public class MapstateMachine extends StateMachine implements Snapshottable,SessionListener {
     private Map<Object, Object> map = new HashMap<>();
+    private Set<ServerSession> sessions = new HashSet<>();
+    private Set<ServerSession> listeners = new HashSet<>();
 
-    public Object put(Commit<PutCommand> commit) {
+    public void listen(Commit<org.tron.consensus.common.Listen> commit) {
+        listeners.add(commit.session());
+        commit.close();
+    }
+
+    public Object put(Commit<org.tron.consensus.common.PutCommand> commit) {
         try {
             map.put(commit.operation().key(), commit.operation().value());
         } finally {
+            commit.session();
             commit.close();
         }
         return null;
     }
 
-    public Object get(Commit<GetQuery> commit) {
+    public Object get(Commit<org.tron.consensus.common.GetQuery> commit) {
         try {
             return map.get(commit.operation().key());
         } finally {
@@ -38,4 +51,29 @@ public void snapshot(SnapshotWriter writer) {
     public void install(SnapshotReader reader) {
         map = reader.readObject();
     }
+
+    /* Listening for Session State Changes */
+    //  called when a new session is registered by a client
+    @Override
+    public void register(ServerSession session) {
+        sessions.add(session);
+    }
+
+    // called when a session is unregistered by a client
+    @Override
+    public void unregister(ServerSession session) {
+
+    }
+
+    // called when a session is expired by the cluster
+    @Override
+    public void expire(ServerSession session) {
+
+    }
+
+    //  called after a session is either unregistered by a client or expired by the leader
+    @Override
+    public void close(ServerSession session) {
+        sessions.remove(session);
+    }
 }