Fix network joining.

Prevent infinite loop in master thread.

6 files changed, 78 insertions, 10 deletions

diff --git a/same/src/main/java/com/orbekk/same/ClientApp.java b/same/src/main/java/com/orbekk/same/ClientApp.java
index 69997ca..bf391dc 100644
--- a/same/src/main/java/com/orbekk/same/ClientApp.java
+++ b/same/src/main/java/com/orbekk/same/ClientApp.java
@@ -42,6 +42,8 @@ public class ClientApp {
             }
         }
         
+        client.joinNetwork(masterUrl + "MasterService.json");
+        
         try {
             server.join();
         } catch (InterruptedException e) {
diff --git a/same/src/main/java/com/orbekk/same/ClientServiceImpl.java b/same/src/main/java/com/orbekk/same/ClientServiceImpl.java
index c648abd..6675c15 100644
--- a/same/src/main/java/com/orbekk/same/ClientServiceImpl.java
+++ b/same/src/main/java/com/orbekk/same/ClientServiceImpl.java
@@ -34,6 +34,16 @@ public class ClientServiceImpl implements ClientService, UrlReceiver {
         return myUrl;
     }
     
+    public void joinNetwork(String masterUrl) {
+        if (myUrl != null) {
+            MasterService master = connections.getMaster(masterUrl);
+            master.joinNetworkRequest(myUrl);          
+        } else {
+            logger.error("Tried to join network at {}, but my url is unknown. " +
+                    "Run discovery service.", masterUrl);
+        }
+    }
+    
     State testGetState() {
         return state;
     }
diff --git a/same/src/main/java/com/orbekk/same/MasterApp.java b/same/src/main/java/com/orbekk/same/MasterApp.java
index c4154da..3e22563 100644
--- a/same/src/main/java/com/orbekk/same/MasterApp.java
+++ b/same/src/main/java/com/orbekk/same/MasterApp.java
@@ -22,6 +22,9 @@ public class MasterApp {
         RpcHandler rpcHandler = new RpcHandler(jsonServer, master);
         server.setHandler(rpcHandler);
         
+        Thread masterThread = new Thread(master);
+        masterThread.start();
+        
         try {
             server.start();
         } catch (Exception e) {
@@ -30,6 +33,7 @@ public class MasterApp {
         
         try {
             server.join();
+            masterThread.join();
         } catch (InterruptedException e) {
             logger.info("Received exception. Exiting. {}", e);
         }
diff --git a/same/src/main/java/com/orbekk/same/MasterServiceImpl.java b/same/src/main/java/com/orbekk/same/MasterServiceImpl.java
index 410def8..729c0ba 100644
--- a/same/src/main/java/com/orbekk/same/MasterServiceImpl.java
+++ b/same/src/main/java/com/orbekk/same/MasterServiceImpl.java
@@ -24,6 +24,7 @@ public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable {
     
     @Override
     public void joinNetworkRequest(String clientUrl) {
+        logger.info("JoinNetworkRequest({})", clientUrl);
         List<String> participants = participants();
         if (!participants.contains(clientUrl)) {
             participants.add(clientUrl);
@@ -53,17 +54,20 @@ public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable {
         return worked;
     }
     
-    public boolean _sendFullState() {
-        boolean worked = _fullStateReceivers.size() != 0;
-        final List<State.Component> components = state.getComponents();
-        broadcaster.broadcast(participants(), new ServiceOperation() {
-            @Override public void run(ClientService client) {
-                for (Component c : components) {
-                    client.setState(c.getName(), c.getData(), c.getRevision());
+    public synchronized boolean _sendFullState() {
+        boolean hasWork = _fullStateReceivers.size() != 0;
+        if (hasWork) {
+            final List<State.Component> components = state.getComponents();
+            broadcaster.broadcast(participants(), new ServiceOperation() {
+                @Override public void run(ClientService client) {
+                    for (Component c : components) {
+                        client.setState(c.getName(), c.getData(), c.getRevision());
+                    }
                 }
-            }
-        });
-        return worked;
+            });
+            _fullStateReceivers.clear();
+        }
+        return hasWork;
     }
     
     private List<String> participants() {
diff --git a/same/src/main/java/com/orbekk/same/TestBroadcaster.java b/same/src/main/java/com/orbekk/same/TestBroadcaster.java
new file mode 100644
index 0000000..434b107
--- /dev/null
+++ b/same/src/main/java/com/orbekk/same/TestBroadcaster.java
@@ -0,0 +1,24 @@
+package com.orbekk.same;
+
+import java.util.List;
+import org.junit.Ignore;
+
+@Ignore
+public class TestBroadcaster implements Broadcaster {
+    public ConnectionManager connections;
+
+    public TestBroadcaster() {
+    }
+
+    public TestBroadcaster(ConnectionManager connections) {
+        this.connections = connections;
+    }
+
+    public void broadcast(final  List<String> targets,
+            final ServiceOperation operation) {
+        for (String t : targets) {
+            ClientService client = connections.getClient(t);
+            operation.run(client);
+        }
+    }
+}
diff --git a/same/src/main/java/com/orbekk/same/TestConnectionManager.java b/same/src/main/java/com/orbekk/same/TestConnectionManager.java
new file mode 100644
index 0000000..a41ec48
--- /dev/null
+++ b/same/src/main/java/com/orbekk/same/TestConnectionManager.java
@@ -0,0 +1,24 @@
+package com.orbekk.same;
+
+import java.util.Map;
+import java.util.HashMap;
+import org.junit.Ignore;
+
+@Ignore
+public class TestConnectionManager implements ConnectionManager {
+    public Map<String, ClientService> clientMap =
+        new HashMap<String, ClientService>();
+    public Map<String, MasterService> masterMap =
+        new HashMap<String, MasterService>();
+
+    public TestConnectionManager() {
+    }
+
+    public ClientService getClient(String url) {
+        return clientMap.get(url);
+    }
+
+    public MasterService getMaster(String url) {
+        return masterMap.get(url);
+    }
+}