diff options
Diffstat (limited to 'same/src/main/java/com')
3 files changed, 60 insertions, 8 deletions
diff --git a/same/src/main/java/com/orbekk/same/BroadcastRunner.java b/same/src/main/java/com/orbekk/same/BroadcastRunner.java new file mode 100644 index 0000000..45935b6 --- /dev/null +++ b/same/src/main/java/com/orbekk/same/BroadcastRunner.java @@ -0,0 +1,33 @@ +package com.orbekk.same; + +import java.util.List; +import java.util.concurrent.Executor; +import java.util.concurrent.Executors; + +public class BroadcastRunner { + private ConnectionManager connections; + private Executor executor; + + /** + * Get a BroadcastRunner for ClientService using a thread pool of size 20. + */ + public static BroadcastRunner getDefaultBroadcastRunner() { + return new BroadcastRunner(Executors.newFixedThreadPool(20)); + } + + public BroadcastRunner(Executor executor, ConnectionManager connections) { + this.executor = executor; + } + + public synchronized void broadcast(final List<String> targets, + final ServiceOperation operation) { + for (String t : targets) { + final ClientService client = connections.getConnection(t); + executor.execute(new Runnable() { + @Override public void run() { + operation.run(client); + } + }); + } + } +} diff --git a/same/src/main/java/com/orbekk/same/MasterServiceImpl.java b/same/src/main/java/com/orbekk/same/MasterServiceImpl.java index 272e405..dcebb80 100644 --- a/same/src/main/java/com/orbekk/same/MasterServiceImpl.java +++ b/same/src/main/java/com/orbekk/same/MasterServiceImpl.java @@ -7,17 +7,21 @@ import org.slf4j.LoggerFactory; public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable { private Logger logger = LoggerFactory.getLogger(getClass()); + private ConnectionManager connections; private State state; private boolean stopped = false; - - public MasterServiceImpl(State initialState) { + private BroadcastRunner broadcaster; + + public MasterServiceImpl(State initialState, ConnectionManager connections, + BroadcastRunner broadcaster) { state = initialState; - } + this.broadcaster = broadcaster; +} @Override public void joinNetworkRequest(String networkName, String clientUrl) { if (networkName.equals(state.getDataOf(".networkName"))) { - List<String> participants = state.getList(".participants"); + List<String> participants = participants(); if (!participants.contains(clientUrl)) { participants.add(clientUrl); synchronized(this) { @@ -35,17 +39,27 @@ public class MasterServiceImpl implements MasterService, UrlReceiver, Runnable { 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)); + for (final String component : state.getAndClearUpdatedComponents()) { + logger.info("Broadcasting new component {}", state.show(component)); + + broadcaster.broadcast(participants(), new ServiceOperation() { + @Override void run(ClientService client) { + client.setState(component, state.getDataOf(component), + state.getRevision(component)); + } + }); worked = true; } return worked; } + private List<String> participants() { + return state.getList(".participants"); + } + + @Override public boolean updateStateRequest(String component, String newData, long revision) { // TODO Auto-generated method stub diff --git a/same/src/main/java/com/orbekk/same/ServiceOperation.java b/same/src/main/java/com/orbekk/same/ServiceOperation.java new file mode 100644 index 0000000..5b99c0b --- /dev/null +++ b/same/src/main/java/com/orbekk/same/ServiceOperation.java @@ -0,0 +1,5 @@ +package com.orbekk.same; + +public interface ServiceOperation { + void run(ClientService service); +} |