diff options
Diffstat (limited to 'src')
5 files changed, 141 insertions, 10 deletions
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/HejkaMessage.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/HejkaMessage.java index 340d939..62554a5 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/HejkaMessage.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/HejkaMessage.java @@ -8,12 +8,16 @@ import pl.edu.mimuw.cloudatlas.model.ValueTime; public class HejkaMessage extends RemoteGossipGirlMessage { private long senderGossipId; + private PathName senderPath; + private PathName receiverPath; private Map<PathName, ValueTime> zoneTimestamps; private Map<Attribute, ValueTime> queryTimestamps; - public HejkaMessage(String messageId, long timestamp, long senderGossipId, Map<PathName, ValueTime> zoneTimestamps, Map<Attribute, ValueTime> queryTimestamps) { + public HejkaMessage(String messageId, long timestamp, long senderGossipId, PathName senderPath, PathName receiverPath, Map<PathName, ValueTime> zoneTimestamps, Map<Attribute, ValueTime> queryTimestamps) { super(messageId, timestamp, Type.HEJKA); this.senderGossipId = senderGossipId; + this.senderPath = senderPath; + this.receiverPath = receiverPath; this.zoneTimestamps = zoneTimestamps; this.queryTimestamps = queryTimestamps; } @@ -22,6 +26,14 @@ public class HejkaMessage extends RemoteGossipGirlMessage { return senderGossipId; } + public PathName getSenderPath() { + return senderPath; + } + + public PathName getReceiverPath() { + return receiverPath; + } + public Map<PathName, ValueTime> getZoneTimestamps() { return zoneTimestamps; } diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/NoCoTamMessage.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/NoCoTamMessage.java index 3dd0c4d..c23722c 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/NoCoTamMessage.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/messages/NoCoTamMessage.java @@ -14,7 +14,7 @@ public class NoCoTamMessage extends RemoteGossipGirlMessage { private ValueTime hejkaSendTimestamp; private ValueTime hejkaReceiveTimestamp; - public NoCoTamMessage(String messageId, long timestamp, long receiverGossipId, long senderGossipId, Map<PathName, ValueTime> zoneTimestamps, Map<Attribute, ValueTime> queryTimestamps, ValueTime hejkaSendTimestamp, ValueTime hejkaReceiveTimestamp) { + public NoCoTamMessage(String messageId, long timestamp, long senderGossipId, long receiverGossipId, Map<PathName, ValueTime> zoneTimestamps, Map<Attribute, ValueTime> queryTimestamps, ValueTime hejkaSendTimestamp, ValueTime hejkaReceiveTimestamp) { super(messageId, timestamp, Type.NO_CO_TAM); this.receiverGossipId = receiverGossipId; this.senderGossipId = senderGossipId; diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java index e7bd227..0c02f1f 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirl.java @@ -20,6 +20,7 @@ import pl.edu.mimuw.cloudatlas.agent.messages.UpdateQueriesMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; import pl.edu.mimuw.cloudatlas.model.AttributesMap; import pl.edu.mimuw.cloudatlas.model.PathName; +import pl.edu.mimuw.cloudatlas.model.ValueContact; import pl.edu.mimuw.cloudatlas.model.ValueQuery; import pl.edu.mimuw.cloudatlas.model.ValueTime; import pl.edu.mimuw.cloudatlas.model.ZMI; @@ -38,6 +39,9 @@ public class GossipGirl extends Module { case INITIATE: initiateGossip((InitiateGossipMessage) message); break; + case HEJKA: + receiveGossip((HejkaMessage) message); + break; case NO_CO_TAM: handleNoCoTam((NoCoTamMessage) message); break; @@ -71,6 +75,23 @@ public class GossipGirl extends Module { sendMessage(getState); } + private void receiveGossip(HejkaMessage message) throws InterruptedException { + Long gossipId = nextGossipId; + nextGossipId++; + gossipStates.put(gossipId, new GossipGirlState( + gossipId, + message.getReceiverPath(), + new ValueContact(message.getSenderPath(), message.getSenderAddress()), + false + ) + ); + + gossipStates.get(gossipId).handleHejka(message); + + GetStateMessage getState = new GetStateMessage("", 0, ModuleType.GOSSIP, gossipId); + sendMessage(getState); + } + private void setState(StateMessage message) throws InterruptedException { GossipGirlState state = gossipStates.get(message.getRequestId()); if (state != null) { @@ -80,12 +101,28 @@ public class GossipGirl extends Module { "", 0, state.gossipId, + state.ourPath, + state.theirContact.getName(), state.getZoneTimestampsToSend(), state.getQueryTimestampsToSend() ); UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, hejka); sendMessage(udupMessage); state.sentHejka(); + } else if (state.state == GossipGirlState.State.SEND_NO_CO_TAM) { + NoCoTamMessage noCoTam = new NoCoTamMessage( + "", + 0, + state.gossipId, + state.theirGossipId, + state.getZoneTimestampsToSend(), + state.getQueryTimestampsToSend(), + state.hejkaSendTimestamp, + state.hejkaReceiveTimestamp + ); + UDUPMessage udupMessage = new UDUPMessage("", 0, state.theirContact, noCoTam); + sendMessage(udupMessage); + state.sentNoCoTam(); } } else { System.out.println("ERROR: GossipGirl got state for a nonexistent gossip"); diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java index 8b0711e..1629914 100644 --- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java +++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlState.java @@ -10,6 +10,7 @@ import java.util.Map; import java.util.Map.Entry; import java.util.Set; +import pl.edu.mimuw.cloudatlas.agent.messages.HejkaMessage; import pl.edu.mimuw.cloudatlas.agent.messages.NoCoTamMessage; import pl.edu.mimuw.cloudatlas.model.Attribute; import pl.edu.mimuw.cloudatlas.model.PathName; @@ -22,8 +23,10 @@ import pl.edu.mimuw.cloudatlas.model.ZMI; public class GossipGirlState { public enum State { WAIT_FOR_STATE_INITIALIZER, + APPLY_HEJKA, WAIT_FOR_STATE_RESPONDER, SEND_HEJKA, + SEND_NO_CO_TAM, SEND_INFO, WAIT_FOR_NO_CO_TAM, WAIT_FOR_FIRST_INFO, @@ -57,7 +60,7 @@ public class GossipGirlState { if (initiating) { state = State.WAIT_FOR_STATE_INITIALIZER; } else { - state = State.WAIT_FOR_STATE_RESPONDER; + state = State.APPLY_HEJKA; } } @@ -71,7 +74,7 @@ public class GossipGirlState { case WAIT_FOR_STATE_RESPONDER: this.hierarchy = hierarchy; this.queries = queries; - state = State.WAIT_FOR_FIRST_INFO; + state = State.SEND_NO_CO_TAM; break; default: System.out.println("ERROR: tried to set gossip state when not expected"); @@ -90,6 +93,34 @@ public class GossipGirlState { } } + public void sentNoCoTam() { + switch (state) { + case SEND_NO_CO_TAM: + state = state.WAIT_FOR_FIRST_INFO; + break; + default: + System.out.println("ERROR: tried to set gossip state when not expected"); + state = State.ERROR; + } + } + + public void handleHejka(HejkaMessage message) { + switch (state) { + case APPLY_HEJKA: + System.out.println("setting sender gossip id to " + message.getSenderGossipId()); + theirGossipId = message.getSenderGossipId(); + theirZoneTimestamps = message.getZoneTimestamps(); + theirQueryTimestamps = message.getQueryTimestamps(); + hejkaSendTimestamp = message.getSentTimestamp(); + hejkaReceiveTimestamp = message.getReceivedTimestamp(); + state = State.WAIT_FOR_STATE_RESPONDER; + break; + default: + System.out.println("ERROR: tried to set gossip state when not expected"); + state = State.ERROR; + } + } + public void handleNoCoTam(NoCoTamMessage message) { switch (state) { case WAIT_FOR_NO_CO_TAM: diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlTest.java index fffb577..170b76e 100644 --- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlTest.java +++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/modules/GossipGirlTest.java @@ -49,13 +49,15 @@ public class GossipGirlTest { private ValueTime testTime; private ZMI initiatorHierarchy; private Map<Attribute, Entry<ValueQuery, ValueTime>> initiatorQueries; - private StateMessage initiatorStateMessage; + private StateMessage stateMessage; private NoCoTamMessage noCoTamMessage; private AttributesMessage attributesMessage1; private AttributesMessage attributesMessage2; private QueryMessage queryMessage1; private QueryMessage queryMessage2; + private HejkaMessage hejkaMessage; + @Before public void setupLocals() throws Exception { gossipGirl = new GossipGirl(); @@ -75,19 +77,25 @@ public class GossipGirlTest { testTime = ValueUtils.currentTime(); setupHierarchy(); setupQueries(); - initiatorStateMessage = new StateMessage("", ModuleType.GOSSIP, 0, 0, initiatorHierarchy, initiatorQueries); + stateMessage = new StateMessage("", ModuleType.GOSSIP, 0, 0, initiatorHierarchy, initiatorQueries); Map<PathName, ValueTime> otherZoneTimestamps = makeOtherZoneTimestamps(); Map<Attribute, ValueTime> otherQueryTimestamps = makeOtherQueryTimestamps(); - noCoTamMessage = new NoCoTamMessage("", 0, 0, 42, otherZoneTimestamps, otherQueryTimestamps, TestUtil.addToTime(testTime, 10), TestUtil.addToTime(testTime, 22)); + noCoTamMessage = new NoCoTamMessage("", 0, 42, 0, otherZoneTimestamps, otherQueryTimestamps, TestUtil.addToTime(testTime, 10), TestUtil.addToTime(testTime, 22)); attributesMessage1 = makeAttributesMessage("/son/bro", makeAttributes1()); attributesMessage2 = makeAttributesMessage("/son/whodis", makeAttributes2()); queryMessage1 = makeQueryMessage("&one", "SELECT 3 AS one"); queryMessage2 = makeQueryMessage("&three", "SELECT 3 AS three"); + + hejkaMessage = new HejkaMessage("", 0, 123, new PathName("/son/bro"), new PathName("/son/grand"), otherZoneTimestamps, otherQueryTimestamps); + hejkaMessage.setSentTimestamp(testTime); + hejkaMessage.setReceivedTimestamp(TestUtil.addToTime(testTime, 15)); + hejkaMessage.setSenderAddress(theirContact.getAddress()); } + public QueryMessage makeQueryMessage(String name, String query) throws Exception { return new QueryMessage("", 0, new Attribute(name), new ValueQuery(query), 0); } @@ -192,7 +200,7 @@ public class GossipGirlTest { public void initiatorSendsHejkaOnState() throws Exception { gossipGirl.handleTyped(initiateGossipMessage); executor.messagesToPass.take(); - gossipGirl.handleTyped(initiatorStateMessage); + gossipGirl.handleTyped(stateMessage); AgentMessage receivedMessage = executor.messagesToPass.poll(); assertUDUPMessage( @@ -202,6 +210,8 @@ public class GossipGirlTest { ); HejkaMessage hejkaMessage = (HejkaMessage) ((UDUPMessage) receivedMessage).getContent(); assertEquals(0, hejkaMessage.getSenderGossipId()); + assertEquals(new PathName("/son/grand"), hejkaMessage.getSenderPath()); + assertEquals(new PathName("/son/bro"), hejkaMessage.getReceiverPath()); assertEquals(3, TestUtil.iterableSize(hejkaMessage.getZoneTimestamps().keySet())); Set<PathName> zoneSet = hejkaMessage.getZoneTimestamps().keySet(); assertThat(zoneSet, hasItems(new PathName("/daughter"))); @@ -219,7 +229,7 @@ public class GossipGirlTest { public void initiatorSendsZonesAndQueriesOnNoCoTam() throws Exception { gossipGirl.handleTyped(initiateGossipMessage); executor.messagesToPass.take(); - gossipGirl.handleTyped(initiatorStateMessage); + gossipGirl.handleTyped(stateMessage); executor.messagesToPass.take(); gossipGirl.handleTyped(noCoTamMessage); @@ -243,7 +253,7 @@ public class GossipGirlTest { public void initiatorModifiesStateOnAttributes() throws Exception { gossipGirl.handleTyped(initiateGossipMessage); executor.messagesToPass.take(); - gossipGirl.handleTyped(initiatorStateMessage); + gossipGirl.handleTyped(stateMessage); executor.messagesToPass.take(); gossipGirl.handleTyped(noCoTamMessage); executor.messagesToPass.take(); @@ -286,6 +296,47 @@ public class GossipGirlTest { gossipGirl.handleTyped(queryMessage2); } + @Test + public void getsStateOnHejka() throws Exception { + gossipGirl.handleTyped(hejkaMessage); + + AgentMessage receivedMessage = executor.messagesToPass.poll(); + assertNotNull(receivedMessage); + assertEquals(ModuleType.STATE, receivedMessage.getDestinationModule()); + StanikMessage stanikMessage = (StanikMessage) receivedMessage; + assertEquals(StanikMessage.Type.GET_STATE, stanikMessage.getType()); + GetStateMessage getStateMessage = (GetStateMessage) stanikMessage; + assertEquals(ModuleType.GOSSIP, getStateMessage.getRequestingModule()); + } + + @Test + public void receiverSendsNoCoTamOnState() throws Exception { + gossipGirl.handleTyped(hejkaMessage); + executor.messagesToPass.poll(); + + gossipGirl.handleTyped(stateMessage); + AgentMessage receivedMessage = executor.messagesToPass.poll(); + assertUDUPMessage( + receivedMessage, + new PathName("/son/bro"), + GossipGirlMessage.Type.NO_CO_TAM + ); + NoCoTamMessage noCoTamMessage = (NoCoTamMessage) ((UDUPMessage) receivedMessage).getContent(); + assertEquals(0, noCoTamMessage.getSenderGossipId()); + assertEquals(123, noCoTamMessage.getReceiverGossipId()); + assertEquals(3, TestUtil.iterableSize(noCoTamMessage.getZoneTimestamps().keySet())); + Set<PathName> zoneSet = noCoTamMessage.getZoneTimestamps().keySet(); + assertThat(zoneSet, hasItems(new PathName("/daughter"))); + assertThat(zoneSet, hasItems(new PathName("/son/sis"))); + assertThat(zoneSet, hasItems(new PathName("/son/grand"))); + + assertEquals(3, TestUtil.iterableSize(noCoTamMessage.getQueryTimestamps().keySet())); + Set<Attribute> querySet = noCoTamMessage.getQueryTimestamps().keySet(); + assertThat(querySet, hasItems(new Attribute("&one"))); + assertThat(querySet, hasItems(new Attribute("&two"))); + assertThat(querySet, hasItems(new Attribute("&query"))); + } + private void assertQueryMessage(AgentMessage message, String recipientPath, String name, String query) throws Exception { assertUDUPMessage( message, |