Merge branch 'master' into gossip-girl-2

4 files changed, 57 insertions, 38 deletions

diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java
index 8c6db8f..501c76e 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java
@@ -23,30 +23,19 @@ import java.util.concurrent.atomic.AtomicBoolean;
  *  due to ValueContact design
  */
 
-// TODO set server port in global config - must be the same everywhere
-// TODO same with buffer size
-
-// TODO separate server like newapiimpl
-// TODO add timestamps as close to sending as possible
-
-// TODO wysylac tylko remotegossipgirl message
-// TODO update timestampow odpowiedni w tym remotegossipgirlmessage
-
-public class UDUP extends Module implements Runnable {
+public class UDUP extends Module {
     private UDUPClient client;
     private UDUPServer server;
-    private final AtomicBoolean running;
 
-    public UDUP(InetAddress serverAddr,
-                int serverPort,
+    public UDUP(int serverPort,
                 int timeout,
-                int bufferSize) {
+                int bufferSize,
+                UDUPServer server) {
         super(ModuleType.UDP);
-        this.running = new AtomicBoolean(false);
         try {
             this.client = new UDUPClient(this, serverPort, bufferSize);
-            this.server = new UDUPServer(this, serverAddr, serverPort, bufferSize);
-            this.running.getAndSet(true);
+            this.server = server;
+            this.server.setUDUP(this);
         } catch (SocketException e) {
             e.printStackTrace();
             this.client.close();
@@ -54,17 +43,8 @@ public class UDUP extends Module implements Runnable {
         }
     }
 
-    public void run() {
-        System.out.println("UDP server running");
-        while(this.running.get()) {
-            try {
-                this.server.acceptMessage();
-            } catch (IOException | InterruptedException e) {
-                e.printStackTrace();
-                this.running.getAndSet(false);
-                this.server.close();
-            }
-        }
+    public UDUPServer getServer() {
+        return this.server;
     }
 
     public void handleTyped(UDUPMessage event) throws InterruptedException {
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java
index 2e4f0b4..089cad2 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPClient.java
@@ -1,6 +1,8 @@
 package pl.edu.mimuw.cloudatlas.agent.modules;
 
 import pl.edu.mimuw.cloudatlas.agent.messages.UDUPMessage;
+import pl.edu.mimuw.cloudatlas.model.ValueTime;
+import pl.edu.mimuw.cloudatlas.model.ValueUtils;
 
 import javax.xml.crypto.Data;
 import java.io.IOException;
@@ -67,9 +69,12 @@ public class UDUPClient {
     public void sendMessage(UDUPMessage msg) throws IOException {
         int packetNo = 1;
         byte[] sendBuf;
-        byte[] dataBuf = this.serializer.serialize(msg);
+        byte[] dataBuf;
         this.lastTransmission++;
 
+        msg.getContent().setSentTimestamp(ValueUtils.currentTime());
+        dataBuf = this.serializer.serialize(msg);
+
         do {
             sendBuf = packSendBuffer(this.lastTransmission, packetNo, dataBuf);
             DatagramPacket packet = new DatagramPacket(sendBuf, 0, sendBuf.length, msg.getContact().getAddress(), this.serverPort);
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPSerializer.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPSerializer.java
index 40c4d7c..0f7b99d 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPSerializer.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPSerializer.java
@@ -12,6 +12,7 @@ import java.io.ByteArrayOutputStream;
 import java.net.Inet4Address;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
+import java.rmi.Remote;
 import java.util.ArrayList;
 import java.util.HashSet;
 import java.util.LinkedHashMap;
@@ -138,6 +139,9 @@ public class UDUPSerializer {
         kryo.register(UDUPMessage.class);
         kryo.register(UpdateAttributesMessage.class);
         kryo.register(UpdateQueriesMessage.class);
+        kryo.register(GossipGirlMessage.class);
+        kryo.register(GossipGirlMessage.Type.class);
+        kryo.register(RemoteGossipGirlMessage.class);
 
         // modules
         kryo.register(TimerScheduledTask.class);
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
index b71a475..94882e4 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
@@ -2,6 +2,7 @@ package pl.edu.mimuw.cloudatlas.agent.modules;
 
 import com.google.common.primitives.Bytes;
 import pl.edu.mimuw.cloudatlas.agent.messages.UDUPMessage;
+import pl.edu.mimuw.cloudatlas.model.ValueUtils;
 
 import java.io.ByteArrayInputStream;
 import java.io.IOException;
@@ -10,22 +11,42 @@ import java.nio.ByteBuffer;
 import java.util.ArrayList;
 import java.util.HashMap;
 import java.util.NoSuchElementException;
+import java.util.concurrent.atomic.AtomicBoolean;
 
-public class UDUPServer {
+public class UDUPServer implements Runnable {
     private UDUP udp;
     private UDUPSerializer serializer;
     private DatagramSocket socket;
     private InetAddress address;
     private HashMap<String, ArrayList<byte[]>> partialPackets;
     private int bufSize;
+    private final AtomicBoolean running;
 
-    public UDUPServer(UDUP udp, InetAddress addr, int port, int bufSize) throws SocketException {
-        this.udp = udp;
+    public UDUPServer(InetAddress addr, int port, int bufSize) throws SocketException {
         this.socket = new DatagramSocket(port, addr);
         this.address = addr;
         this.bufSize = bufSize;
         this.partialPackets = new HashMap<>();
         this.serializer = new UDUPSerializer();
+        this.running = new AtomicBoolean(false);
+    }
+
+    public void setUDUP(UDUP udup) {
+        this.udp = udup;
+    }
+
+    public void run() {
+        System.out.println("UDP server running");
+        this.running.getAndSet(true);
+        while(this.running.get()) {
+            try {
+                this.acceptMessage();
+            } catch (IOException | InterruptedException e) {
+                e.printStackTrace();
+                this.running.getAndSet(false);
+                this.close();
+            }
+        }
     }
 
     public void acceptMessage() throws IOException, InterruptedException {
@@ -34,14 +55,19 @@ public class UDUPServer {
         String transmissionID = packTransmissionID(transmissionNo, packet.getAddress());
         int packetNo = readPacketNo(packet.getData());
         byte[] packetData = trimPacketBuffer(packet.getData());
+        UDUPMessage msg;
 
         if (packetNo == 1 && packet.getLength() < this.bufSize) {
-            UDUPMessage msg = this.serializer.deserialize(packetData);
+            msg = this.serializer.deserialize(packetData);
+            msg.getContent().setReceivedTimestamp(ValueUtils.currentTime());
             System.out.println("UDP received message " + msg.getContent().getMessageId());
-            sendMessageFurther(msg);
         } else {
             System.out.println("UDP received partial message with transmission id " + transmissionID + " packet no " + packetNo);
-            this.addPartialMessageAndCheckSerialization(transmissionID, packetNo, packetData);
+            msg = this.addPartialMessageAndCheckSerialization(transmissionID, packetNo, packetData);
+        }
+
+        if (msg != null) {
+            sendMessageFurther(msg);
         }
     }
 
@@ -104,14 +130,16 @@ public class UDUPServer {
         return fullData;
     }
 
-    public void addPartialMessageAndCheckSerialization(String transmissionID, int newPacketNo, byte[] packetData) {
+    public UDUPMessage addPartialMessageAndCheckSerialization(String transmissionID, int newPacketNo, byte[] packetData) {
+        UDUPMessage msg = null;
+
         if (this.partialPackets.containsKey(transmissionID)) {
             try {
                 byte[] allPacketData = concatPacketData(transmissionID, newPacketNo, packetData);
-                UDUPMessage msg = this.serializer.deserialize(allPacketData);
+                msg = this.serializer.deserialize(allPacketData);
+                msg.getContent().setReceivedTimestamp(ValueUtils.currentTime());
                 this.partialPackets.remove(transmissionID);
                 System.out.println("Kryo put together whole transmission for msg " + msg.getContent().getMessageId());
-                this.udp.sendMessage(msg.getContent());
             } catch (Error | Exception e) {
                 System.out.println("Kryo didn't deserialize partial message, waiting to receive the rest");
             }
@@ -120,6 +148,8 @@ public class UDUPServer {
             newTransmission.add(newPacketNo-1, packetData);
             this.partialPackets.put(transmissionID, newTransmission);
         }
+
+        return msg;
     }
 
     public void close() {