Send NoCoTam when responder gets state

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,