Fix handling of multipart udp messages

2 files changed, 110 insertions, 74 deletions

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 6807a86..fb79dc6 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
@@ -31,12 +31,11 @@ public class UDUPServer {
         System.out.println("UDP " + this.address + " received packet from " + packet.getAddress());
 
         if (packet.getOffset() == 0) {
-            UDUPMessage msg = this.serializer.deserialize(packet.getData());
-            System.out.println("UDP received message " + msg.getContent().getMessageId());
-
             if (packet.getLength() == this.bufSize) {
                 this.addPartialMessageAndCheckSerialization(packet.getAddress(), packet.getData());
             } else  {
+                UDUPMessage msg = this.serializer.deserialize(packet.getData());
+                System.out.println("UDP received message " + msg.getContent().getMessageId());
                 if (msg.getContent().getDestinationModule() == ModuleType.TEST) {
                     System.out.println("UDP server: test message received");
                 } else if (msg.getContent().getDestinationModule() != ModuleType.UDP) {
diff --git a/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java b/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
index e4f601a..c4bb8e7 100644
--- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
+++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
@@ -2,11 +2,14 @@ package pl.edu.mimuw.cloudatlas.agent;
 
 import org.junit.*;
 import pl.edu.mimuw.cloudatlas.agent.messages.UDUPMessage;
+import pl.edu.mimuw.cloudatlas.agent.messages.UpdateAttributesMessage;
 import pl.edu.mimuw.cloudatlas.agent.modules.Module;
 import pl.edu.mimuw.cloudatlas.agent.modules.ModuleType;
 import pl.edu.mimuw.cloudatlas.agent.modules.UDUP;
+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.ValueInt;
 
 import java.net.InetAddress;
 import java.net.UnknownHostException;
@@ -78,45 +81,64 @@ public class UDUPTest {
             Assert.fail();
         }
     }
-/*
+
     @Test
     public void bigMessageBetweenUDUPs() {
-        UDUP udp1 = new UDUP(
-                ModuleType.UDP,
-                5999,
-                1000,
-                5000,
-                256);
-
-        UDUP udp2 = new UDUP(
-                ModuleType.UDP,
-                5998,
-                1000,
-                5000,
-                256);
-
-        AttributesMap bigAttrib = new AttributesMap();
-        for (Long i = 1L; i < 20; i++) {
-            bigAttrib.add(i.toString(), new ValueInt(i));
+        UDUP udp1 = null;
+        UDUP udp2 = null;
+        UDUPMessage msg1 = null;
+        boolean testSuccess = true;
+        int timeout = 5000;
+
+        try {
+            System.out.println("Starting udp1");
+
+            udp1 = new UDUP(
+                    InetAddress.getByName("127.0.0.2"),
+                    5998,
+                    timeout,
+                    20);
+
+            System.out.println("Starting udp2");
+
+            udp2 = new UDUP(
+                    InetAddress.getByName("127.0.0.3"),
+                    5998,
+                    timeout,
+                    20);
+
+            UDUPMessage testContent = new UDUPMessage();
+            testContent.setDestinationModule(ModuleType.TEST);
+
+            msg1 = new UDUPMessage(
+                    "udup1",
+                    new ValueContact(new PathName("/udp2"), InetAddress.getByName("127.0.0.3")),
+                    testContent
+            );
+
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
         }
 
-        UDUPMessage msg1 = new UDUPMessage(
-                "udp1",
-                ModuleType.UDP,
-                "localhost",
-                5998,
-                new UpdateAttributesMessage("updateattrib1", 0,"/", bigAttrib),
-                0,
-                "conv1");
+        Thread udpThread1 = new Thread(udp1);
+        udpThread1.start();
+        Thread udpThread2 = new Thread(udp2);
+        udpThread2.start();
 
         try {
             udp1.handle(msg1);
-            Thread.sleep(1000);
-            UDUPMessage conv = udp2.fetchConversation("conv1");
-            Assert.assertEquals(conv.getConversationId(), "conv1");
+            Thread.sleep(timeout + 1000);
         } catch (InterruptedException | Module.InvalidMessageType e) {
             e.printStackTrace();
-        } catch (UDUP.InvalidConversation noConversationError) {
+            testSuccess = false;
+        }
+
+        udpThread1.interrupt();
+        udpThread2.interrupt();
+
+        if (testSuccess) {
+            Assert.assertTrue(true);
+        } else {
             Assert.fail();
         }
     }
@@ -124,53 +146,68 @@ public class UDUPTest {
 
     @Test
     public void sendMultipleMessages() {
-        @Test
-        public void messageBetweenUDUPs() {
-            UDUP udp1 = new UDUP(
-                    ModuleType.UDP,
-                    5999,
-                    1000,
-                    5000,
-                    3,
-                    256);
+        UDUP udp1 = null;
+        UDUP udp2 = null;
+        UDUPMessage msg1 = null;
+        UDUPMessage msg2 = null;
+        UDUPMessage msg3 = null;
+        boolean testSuccess = true;
+        int timeout = 5000;
 
-            UDUP udp2 = new UDUP(
-                    ModuleType.UDP,
-                    5998,
-                    1000,
-                    5000,
-                    3,
-                    256);
+        try {
+            System.out.println("Starting udp1");
 
-            UDUPMessage msg1 = new UDUPMessage(
-                    "1",
-                    ModuleType.UDP,
-                    "localhost",
-                    5998,
-                    null,
-                    0,
-                    "conv1");
+            udp1 = new UDUP(
+                    InetAddress.getByName("127.0.0.2"),
+                    5997,
+                    timeout,
+                    1000);
 
+            System.out.println("Starting udp2");
 
+            udp2 = new UDUP(
+                    InetAddress.getByName("127.0.0.3"),
+                    5997,
+                    timeout,
+                    1000);
 
+            UDUPMessage testContent = new UDUPMessage();
+            testContent.setDestinationModule(ModuleType.TEST);
 
-            try {
-                udp1.handle(msg1);
-                udp1.handle(msg2);
-                udp1.handle(msg3);
-                Thread.sleep(1000);
-                UDUPMessage conv1 = udp2.fetchConversation("conv1");
-                Assert.assertEquals(conv1.getConversationId(), "conv1");
-                UDUPMessage conv2 = udp2.fetchConversation("conv2");
-                Assert.assertEquals(conv2.getConversationId(), "conv2");
-                UDUPMessage conv3 = udp2.fetchConversation("conv3");
-                Assert.assertEquals(conv3.getConversationId(), "conv3");
-            } catch (InterruptedException | Module.InvalidMessageType e) {
-                e.printStackTrace();
-            } catch (UDUP.InvalidConversation invalidConversation) {
-                Assert.fail();
-            }
+            msg1 = new UDUPMessage(
+                    "udup1",
+                    new ValueContact(new PathName("/udp2"), InetAddress.getByName("127.0.0.3")),
+                    testContent
+            );
+
+            msg2 = new UDUPMessage(
+                    "udup2",
+                    new ValueContact(new PathName("/udp2"), InetAddress.getByName("127.0.0.3")),
+                    testContent
+            );
+
+            msg3 = new UDUPMessage(
+                    "udup3",
+                    new ValueContact(new PathName("/udp2"), InetAddress.getByName("127.0.0.3")),
+                    testContent
+            );
+
+        } catch (UnknownHostException e) {
+            e.printStackTrace();
+        }
+
+        Thread udpThread1 = new Thread(udp1);
+        udpThread1.start();
+        Thread udpThread2 = new Thread(udp2);
+        udpThread2.start();
+
+        try {
+            udp1.handle(msg1);
+            udp1.handle(msg2);
+            udp1.handle(msg3);
+            Thread.sleep(timeout + 2000);
+        } catch (InterruptedException | Module.InvalidMessageType e) {
+            e.printStackTrace();
         }
     }
-*/
 }