Bring back worker thread.

:(

3 files changed, 56 insertions, 20 deletions

diff --git a/same/src/main/java/com/orbekk/same/MasterServiceImpl.java b/same/src/main/java/com/orbekk/same/MasterServiceImpl.java
index 50a5607..272e405 100644
--- a/same/src/main/java/com/orbekk/same/MasterServiceImpl.java
+++ b/same/src/main/java/com/orbekk/same/MasterServiceImpl.java
@@ -5,20 +5,24 @@ import java.util.List;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-public class MasterServiceImpl implements MasterService, UrlReceiver {
+public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable {
     private Logger logger = LoggerFactory.getLogger(getClass());
     private State state;
+    private boolean stopped = false;
   
     public MasterServiceImpl(State initialState) {
         state = initialState;
     }
     
     @Override
-    public synchronized void joinNetworkRequest(String networkName, String clientUrl) {
+    public void joinNetworkRequest(String networkName, String clientUrl) {
         if (networkName.equals(state.getDataOf(".networkName"))) {
             List<String> participants = state.getList(".participants");
             if (!participants.contains(clientUrl)) {
                 participants.add(clientUrl);
+                synchronized(this) {
+                    notifyAll();
+                }
             } else {                
                 logger.warn("Client {} already part of network. " +
                         "Ignoring participation request", clientUrl);
@@ -31,7 +35,17 @@ public class MasterServiceImpl implements MasterService, UrlReceiver {
                             state.getDataOf(".networkName") });
         }
     }
-
+    
+    public boolean _handleJoinNetworkRequests() {
+        boolean worked = false;
+        for (String component : state.getAndClearUpdatedComponents()) {
+            logger.error("TODO: Send state update for component {}",
+                    state.show(component));
+            worked = true;
+        }
+        return worked;
+    }
+    
     @Override
     public boolean updateStateRequest(String component, String newData, long revision) {
         // TODO Auto-generated method stub
@@ -44,4 +58,25 @@ public class MasterServiceImpl implements MasterService, UrlReceiver {
         logger.info("Master URL is " + myUrl);
         state.update(".masterUrl", myUrl, 0);
     }
+
+    boolean _performWork() {
+        boolean worked = false;
+        worked |= _handleJoinNetworkRequests();
+        return worked;
+    }
+    
+    @Override
+    public void run() {
+        while (!stopped) {
+            if (_performWork()) {
+                synchronized (this) {
+                    try {
+                        wait(500);
+                    } catch (InterruptedException e) {
+                        // Ignore interrupt in wait loop.
+                    }
+                }
+            }
+        }
+    }
 }
diff --git a/same/src/main/java/com/orbekk/same/State.java b/same/src/main/java/com/orbekk/same/State.java
index 49ee8c4..afd93f5 100644
--- a/same/src/main/java/com/orbekk/same/State.java
+++ b/same/src/main/java/com/orbekk/same/State.java
@@ -17,18 +17,7 @@ import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
 /**
- * This class is not thread-safe. In particular, modifications to this class
- * while getUpdatedComponents() is being used is unsafe. When accessing
- * updatedComponents, synchronize on State:
- * 
- * <code>
- * synchronize (state) {
- *   for (String component : state.updatedComponents()) {
- *     // Do work.
- *   }
- *   state.clearUpdatedComponents();
- * }
- * </code>
+ * This class is thread-safe.
  */
 public class State {
     private Logger logger = LoggerFactory.getLogger(getClass());
@@ -128,12 +117,17 @@ public class State {
         }
     }
     
-    public Set<String> getUpdatedComponents() {
-        return updatedComponents;
+    /**
+     * Pretty print a component.
+     */
+    public String show(String componentName) {
+        return componentName + ": " + state.get(componentName);
     }
-
-    public void clearUpdatedComponents() {
-        this.updatedComponents.clear();
+    
+    public synchronized Set<String> getAndClearUpdatedComponents() {
+        Set<String> copy = new TreeSet<String>(updatedComponents);
+        updatedComponents.clear();
+        return copy;
     }
 
     public static class Component {
diff --git a/same/src/test/java/com/orbekk/same/MasterServiceImplTest.java b/same/src/test/java/com/orbekk/same/MasterServiceImplTest.java
index 9d8a134..cbfbf12 100644
--- a/same/src/test/java/com/orbekk/same/MasterServiceImplTest.java
+++ b/same/src/test/java/com/orbekk/same/MasterServiceImplTest.java
@@ -34,4 +34,11 @@ public class MasterServiceImplTest {
         List<String> participants = state.getList(".participants");
         assertTrue(participants.contains("http://clientUrl"));
     }
+    
+    @Test
+    public void workLoopClearsUpdatedComponents() {
+        state.update("Test", "Content", 0);
+        assertTrue(master._performWork());
+        assertTrue(state.getAndClearUpdatedComponents().isEmpty());
+    }
 }