Add participant synchronization.

- No error handling in the protocol at the moment.

4 files changed, 102 insertions, 20 deletions

diff --git a/jsonrpc/src/main/java/com/orbekk/same/SameService.java b/jsonrpc/src/main/java/com/orbekk/same/SameService.java
index 9e1071d..8f239da 100644
--- a/jsonrpc/src/main/java/com/orbekk/same/SameService.java
+++ b/jsonrpc/src/main/java/com/orbekk/same/SameService.java
@@ -14,12 +14,27 @@ public interface SameService {
     /**
      * A request from the callee to participate in 'networkName'.
      */
-    void participateNetwork(String networkName, String clientId,
-            String url);
+    void participateNetwork(String networkName, String clientId, String url);
 
     /**
      * Notification of participation in network.
      */
-    void notifyParticipation(String networkName, String masterId,
-            Map<String, String> participants);
+    void notifyParticipation(String networkName, String masterId);
+
+    /**
+     * New state.
+     * 
+     * When sent to a non-master from the master, use 'newState' as the
+     * current state.
+     *
+     * When sent to a master, broadcast the new state to all clients.
+     */
+    void setState(String newState);
+
+    /**
+     * Notify all nodes of network participants.
+     *
+     * Only sent from master to non-master.
+     */
+    void setParticipants(Map<String, String> participants);
 }
diff --git a/jsonrpc/src/main/java/com/orbekk/same/SameServiceImpl.java b/jsonrpc/src/main/java/com/orbekk/same/SameServiceImpl.java
index 0e17f54..a946c61 100644
--- a/jsonrpc/src/main/java/com/orbekk/same/SameServiceImpl.java
+++ b/jsonrpc/src/main/java/com/orbekk/same/SameServiceImpl.java
@@ -35,17 +35,26 @@ public class SameServiceImpl implements SameService {
     }
 
     @Override
-    public void notifyParticipation(String networkName, String masterId,
-            Map<String, String> participants) {
+    public void notifyParticipation(String networkName, String masterId) {
         logger.info("Joining network {}. Master is {}", networkName, masterId);
-        int i = 1;
-        for (Map.Entry<String, String> e : participants.entrySet()) {
-            String clientId = e.getKey();
-            String url = e.getValue();
-            logger.info("  {} participant {}: {}, {}",
-                    new Object[]{networkName, i, clientId, url});
-            i++;
-        }
-        sameState.joinNetwork(networkName, masterId, participants);
+        // int i = 1;
+        // for (Map.Entry<String, String> e : participants.entrySet()) {
+        //     String clientId = e.getKey();
+        //     String url = e.getValue();
+        //     logger.info("  {} participant {}: {}, {}",
+        //             new Object[]{networkName, i, clientId, url});
+        //     i++;
+        // }
+        sameState.joinNetwork(networkName, masterId);
+    }
+
+    @Override
+    public void setParticipants(Map<String, String> participants) {
+        sameState.setParticipants(participants);
+    }
+
+    @Override
+    public void setState(String newState) {
+        logger.error("setState not implemented.");
     }
 }
diff --git a/jsonrpc/src/main/java/com/orbekk/same/SameState.java b/jsonrpc/src/main/java/com/orbekk/same/SameState.java
index 9638252..d44cc1d 100644
--- a/jsonrpc/src/main/java/com/orbekk/same/SameState.java
+++ b/jsonrpc/src/main/java/com/orbekk/same/SameState.java
@@ -38,6 +38,13 @@ public class SameState extends Thread implements UrlReceiver {
     private Map<String, String> participants = new HashMap<String, String>();
 
     /**
+     * New participants.
+     *
+     * New participants map to replace participants.
+     */
+    private Map<String, String> _setParticipants = null;
+
+    /**
      * The client id of this participant.
      */
     private String clientId;
@@ -74,15 +81,16 @@ public class SameState extends Thread implements UrlReceiver {
      * TODO: Implement fully.
      */
     private synchronized void resetState() {
+        networkName = "";
+        masterId = "";
         pendingParticipants.clear();
+        participants.clear();
     }
 
-    public synchronized void joinNetwork(String networkName, String masterId,
-            Map<String, String> participants) {
+    public synchronized void joinNetwork(String networkName, String masterId) {
         resetState();
         this.networkName = networkName;
         this.masterId = masterId;
-        this.participants = participants;
         logger.info("Joined network {}.", networkName);
     }
 
@@ -114,6 +122,24 @@ public class SameState extends Thread implements UrlReceiver {
         notifyAll();
     }
 
+    public synchronized void setParticipants(Map<String, String> participants) {
+        logger.info("Pending operation: _setParticipants");
+        _setParticipants = participants;
+        notifyAll();
+    }
+
+    private synchronized void handleSetParticipants() {
+        if (_setParticipants != null) {
+            if (isMaster()) {
+                logger.error("{}: Master received setParticipants.", clientId);
+            } else {
+                logger.info("{}: New participants committed.", clientId);
+                participants = _setParticipants;
+            }
+        }
+        _setParticipants = null;
+    }
+
     private boolean isMaster() {
         return masterId.equals(clientId);
     }
@@ -136,8 +162,12 @@ public class SameState extends Thread implements UrlReceiver {
                 String url = e.getValue();
                 logger.info("New participant: {} URL({})", clientId, url);
                 SameService remoteService = connections.getConnection(url);
-                remoteService.notifyParticipation(networkName, masterId,
-                        participants);
+                remoteService.notifyParticipation(networkName, masterId);
+                broadcast(new ServiceOperation(){
+                    @Override void run(SameService service) {
+                        service.setParticipants(participants);
+                    }
+                });
             }
         }
         pendingParticipants.clear();
@@ -151,6 +181,7 @@ public class SameState extends Thread implements UrlReceiver {
      */
     synchronized void internalRun() {
         handleNewParticipants();
+        handleSetParticipants();
     }
 
     public synchronized void run() {
@@ -174,4 +205,18 @@ public class SameState extends Thread implements UrlReceiver {
                     "to finish. Ignoring.");
         }
     }
+
+    public abstract static class ServiceOperation {
+        abstract void run(SameService service);
+    }
+    
+    public synchronized void broadcast(ServiceOperation operation) {
+        for (Map.Entry<String, String> e : participants.entrySet()) {
+            String clientId = e.getKey();
+            String url = e.getValue();
+            if (!clientId.equals(this.clientId)) {
+                operation.run(connections.getConnection(url));
+            }
+        }
+    }
 }
diff --git a/jsonrpc/src/test/java/com/orbekk/same/SameStateTest.java b/jsonrpc/src/test/java/com/orbekk/same/SameStateTest.java
index 77bbbee..7e21e7a 100644
--- a/jsonrpc/src/test/java/com/orbekk/same/SameStateTest.java
+++ b/jsonrpc/src/test/java/com/orbekk/same/SameStateTest.java
@@ -56,5 +56,18 @@ public class SameStateTest {
         assertTrue(state1.getParticipants().size() == 2);
         assertTrue(state2.getParticipants().size() == 2);
         assertEquals(state1.getNetworkName(), state2.getNetworkName());
+
+        connections.getConnection(state2.getUrl()).
+            participateNetwork("Network1", state3.getClientId(),
+                    state3.getUrl());
+        state2.internalRun();
+        state1.internalRun();
+        state3.internalRun();
+
+        assertTrue(state1.getParticipants().size() == 3);
+        assertTrue(state2.getParticipants().size() == 3);
+        assertTrue(state3.getParticipants().size() == 3);
+        assertEquals(state1.getNetworkName(), state2.getNetworkName());
+        assertEquals(state2.getNetworkName(), state3.getNetworkName());
     }
 }