From 0822aaaeef7f9d3050fc3212d0c33d1878d75833 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Kjetil=20=C3=98rbekk?= Date: Tue, 17 Apr 2012 14:15:08 +0200 Subject: Add new participants list. The new list, '.participants0' will hold the location of the protobuf services of clients. --- same/src/main/java/com/orbekk/same/App.java | 2 +- same/src/main/java/com/orbekk/same/Master.java | 30 +++++++++++++++++++++++++- same/src/main/java/com/orbekk/same/State.java | 3 +++ 3 files changed, 33 insertions(+), 2 deletions(-) diff --git a/same/src/main/java/com/orbekk/same/App.java b/same/src/main/java/com/orbekk/same/App.java index 2b2985e..cc8b9ce 100644 --- a/same/src/main/java/com/orbekk/same/App.java +++ b/same/src/main/java/com/orbekk/same/App.java @@ -10,7 +10,7 @@ public class App { private Services.MasterState getMasterInfo(Configuration configuration) { return Services.MasterState.newBuilder() - .setMasterUrl(configuration.get(".masterUrl")) + .setMasterUrl(configuration.get("masterUrl")) .setMasterLocation(configuration.get("masterLocation")) .build(); } diff --git a/same/src/main/java/com/orbekk/same/Master.java b/same/src/main/java/com/orbekk/same/Master.java index d1f27e9..b2adea1 100644 --- a/same/src/main/java/com/orbekk/same/Master.java +++ b/same/src/main/java/com/orbekk/same/Master.java @@ -23,7 +23,7 @@ public class Master { State state; private Broadcaster broadcaster; private volatile int masterId = 1; - + public static Master create(ConnectionManager connections, Broadcaster broadcaster, String myUrl, String networkName, String myLocation) { @@ -67,6 +67,8 @@ public class Master { @Override public void joinNetworkRequest(RpcController controller, ClientState request, RpcCallback done) { logger.info("joinNetworkRequest({})", request); + + /** Old participant code. */ List participants = state.getList(".participants"); sendFullStateThread.add(request.getUrl()); if (!participants.contains(request.getUrl())) { @@ -79,9 +81,14 @@ public class Master { } else { logger.warn("Client {} joining: Already part of network"); } + + /** New participant code. */ + addParticipant(request); + done.run(Empty.getDefaultInstance()); } + @Override public void updateStateRequest(RpcController controller, Services.Component request, RpcCallback done) { @@ -192,7 +199,19 @@ public class Master { return serviceImpl; } + private synchronized void addParticipant(ClientState client) { + List participants = state.getList(State.PARTICIPANTS); + if (!participants.contains(client.getUrl())) { + participants.add(client.getUrl()); + state.updateFromObject(State.PARTICIPANTS, participants, + state.getRevision(State.PARTICIPANTS) + 1); + updateStateRequestThread.add(State.PARTICIPANTS); + } + + } + private synchronized void removeParticipant(String url) { + /** TODO: Remove this code. */ List participants = state.getList(".participants"); if (participants.contains(url)) { logger.info("removeParticipant({})", url); @@ -201,6 +220,15 @@ public class Master { state.getRevision(".participants") + 1); updateStateRequestThread.add(".participants"); } + + List participants0 = state.getList(State.PARTICIPANTS); + if (participants0.contains(url)) { + logger.info("removeParticipant({})", url); + participants0.remove(url); + state.updateFromObject(State.PARTICIPANTS, participants0, + state.getRevision(State.PARTICIPANTS) + 1); + updateStateRequestThread.add(State.PARTICIPANTS); + } } private void broadcastNewComponents(List destinations, diff --git a/same/src/main/java/com/orbekk/same/State.java b/same/src/main/java/com/orbekk/same/State.java index f0fee80..9d23d78 100644 --- a/same/src/main/java/com/orbekk/same/State.java +++ b/same/src/main/java/com/orbekk/same/State.java @@ -25,9 +25,12 @@ public class State { private ObjectMapper mapper = new ObjectMapper(); private Set updatedComponents = new TreeSet(); + public static final String PARTICIPANTS = ".participants0"; + public State(String networkName) { update(".networkName", networkName, 1); updateFromObject(".participants", new ArrayList(), 1); + updateFromObject(PARTICIPANTS, new ArrayList(), 1); } public State(State other) { -- cgit v1.2.3