m-chrzan.xyz
aboutsummaryrefslogtreecommitdiff
path: root/src
diff options
context:
space:
mode:
Diffstat (limited to 'src')
-rw-r--r--src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java23
-rw-r--r--src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUP.java27
-rw-r--r--src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java26
-rw-r--r--src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java69
4 files changed, 88 insertions, 57 deletions
diff --git a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java
index 256fa6b..ad2a650 100644
--- a/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java
+++ b/src/main/java/pl/edu/mimuw/cloudatlas/agent/Agent.java
@@ -39,18 +39,15 @@ public class Agent {
}
}
- public static HashMap<ModuleType, Module> initializeModules() {
+ public static HashMap<ModuleType, Module> initializeModules() throws UnknownHostException {
HashMap<ModuleType, Module> modules = new HashMap<ModuleType, Module>();
modules.put(ModuleType.TIMER_SCHEDULER, new TimerScheduler(ModuleType.TIMER_SCHEDULER));
modules.put(ModuleType.RMI, new Remik());
Long freshnessPeriod = new Long(System.getProperty("freshness_period"));
modules.put(ModuleType.STATE, new Stanik(freshnessPeriod));
modules.put(ModuleType.QUERY, new Qurnik());
- try {
- modules.put(ModuleType.UDP, new UDUP(InetAddress.getByName("127.0.0.1"), 5988, 5000, 20000));
- } catch (UnknownHostException e) {
- e.printStackTrace();
- }
+ UDUPServer server = new UDUPServer(InetAddress.getByName("127.0.0.1"), 5988, 2000);
+ modules.put(ModuleType.UDP, new UDUP(5988, 5000, 20000, null));
// TODO add modules as we implement them
return modules;
}
@@ -89,14 +86,24 @@ public class Agent {
}
public static void runModulesAsThreads() {
- HashMap<ModuleType, Module> modules = initializeModules();
+ HashMap<ModuleType, Module> modules = null;
+
+ try {
+ modules = initializeModules();
+ } catch (UnknownHostException e) {
+ System.out.println("Module initialization failed");
+ e.printStackTrace();
+ return;
+ }
+
HashMap<ModuleType, Executor> executors = initializeExecutors(modules);
ArrayList<Thread> executorThreads = initializeExecutorThreads(executors);
-
eventBus = new EventBus(executors);
+ Thread UDUPServerThread = new Thread(((UDUP) modules.get(ModuleType.UDP)).getServer());
Thread eventBusThread = new Thread(eventBus);
System.out.println("Initializing event bus");
eventBusThread.start();
+ UDUPServerThread.start();
}
private static void initZones() {
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..341d947 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
@@ -32,21 +32,19 @@ import java.util.concurrent.atomic.AtomicBoolean;
// 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 +52,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/UDUPServer.java b/src/main/java/pl/edu/mimuw/cloudatlas/agent/modules/UDUPServer.java
index b71a475..d6180be 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
@@ -10,22 +10,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 {
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 13a322b..93ed8be 100644
--- a/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
+++ b/src/test/java/pl/edu/mimuw/cloudatlas/agent/UDUPTest.java
@@ -6,12 +6,14 @@ 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.agent.modules.UDUPServer;
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.SocketException;
import java.net.UnknownHostException;
// TODO add serialization tests that target custom serializers (type collections!)
@@ -22,25 +24,30 @@ public class UDUPTest {
public void messageBetweenUDUPs() {
UDUP udp1 = null;
UDUP udp2 = null;
+ UDUPServer server1 = null;
+ UDUPServer server2 = null;
UDUPMessage msg1 = null;
boolean testSuccess = true;
+ int timeout = 5000;
try {
System.out.println("Starting udp1");
+ server1 = new UDUPServer(InetAddress.getByName("127.0.0.2"), 5996, 1000);
udp1 = new UDUP(
- InetAddress.getByName("127.0.0.2"),
- 5999,
- 5000,
- 1000);
+ 5997,
+ timeout,
+ 1000,
+ server1);
System.out.println("Starting udp2");
+ server2 = new UDUPServer(InetAddress.getByName("127.0.0.3"), 5996, 1000);
udp2 = new UDUP(
- InetAddress.getByName("127.0.0.3"),
- 5999,
- 5000,
- 1000);
+ 5997,
+ timeout,
+ 1000,
+ server2);
UDUPMessage testContent = new UDUPMessage();
testContent.setDestinationModule(ModuleType.TEST);
@@ -52,13 +59,13 @@ public class UDUPTest {
testContent
);
- } catch (UnknownHostException e) {
+ } catch (UnknownHostException | SocketException e) {
e.printStackTrace();
}
- Thread udpThread1 = new Thread(udp1);
+ Thread udpThread1 = new Thread(server1);
udpThread1.start();
- Thread udpThread2 = new Thread(udp2);
+ Thread udpThread2 = new Thread(server2);
udpThread2.start();
try {
@@ -89,6 +96,8 @@ public class UDUPTest {
public void bigMessageBetweenUDUPs() {
UDUP udp1 = null;
UDUP udp2 = null;
+ UDUPServer server1 = null;
+ UDUPServer server2 = null;
UDUPMessage msg1 = null;
boolean testSuccess = true;
int timeout = 5000;
@@ -96,19 +105,21 @@ public class UDUPTest {
try {
System.out.println("Starting udp1");
+ server1 = new UDUPServer(InetAddress.getByName("127.0.0.2"), 5991, 1000);
udp1 = new UDUP(
- InetAddress.getByName("127.0.0.2"),
- 5998,
+ 5997,
timeout,
- 30);
+ 30,
+ server1);
System.out.println("Starting udp2");
+ server2 = new UDUPServer(InetAddress.getByName("127.0.0.3"), 5991, 1000);
udp2 = new UDUP(
- InetAddress.getByName("127.0.0.3"),
- 5998,
+ 5997,
timeout,
- 30);
+ 30,
+ server2);
UDUPMessage testContent = new UDUPMessage();
testContent.setDestinationModule(ModuleType.TEST);
@@ -120,13 +131,13 @@ public class UDUPTest {
testContent
);
- } catch (UnknownHostException e) {
+ } catch (UnknownHostException | SocketException e) {
e.printStackTrace();
}
- Thread udpThread1 = new Thread(udp1);
+ Thread udpThread1 = new Thread(server1);
udpThread1.start();
- Thread udpThread2 = new Thread(udp2);
+ Thread udpThread2 = new Thread(server2);
udpThread2.start();
try {
@@ -152,6 +163,8 @@ public class UDUPTest {
public void sendMultipleMessages() {
UDUP udp1 = null;
UDUP udp2 = null;
+ UDUPServer server1 = null;
+ UDUPServer server2 = null;
UDUPMessage msg1 = null;
UDUPMessage msg2 = null;
UDUPMessage msg3 = null;
@@ -161,19 +174,21 @@ public class UDUPTest {
try {
System.out.println("Starting udp1");
+ server1 = new UDUPServer(InetAddress.getByName("127.0.0.2"), 5997, 1000);
udp1 = new UDUP(
- InetAddress.getByName("127.0.0.2"),
5997,
timeout,
- 1000);
+ 1000,
+ server1);
System.out.println("Starting udp2");
+ server2 = new UDUPServer(InetAddress.getByName("127.0.0.3"), 5997, 1000);
udp2 = new UDUP(
- InetAddress.getByName("127.0.0.3"),
5997,
timeout,
- 1000);
+ 1000,
+ server2);
UDUPMessage testContent = new UDUPMessage();
testContent.setDestinationModule(ModuleType.TEST);
@@ -197,13 +212,13 @@ public class UDUPTest {
testContent
);
- } catch (UnknownHostException e) {
+ } catch (UnknownHostException | SocketException e) {
e.printStackTrace();
}
- Thread udpThread1 = new Thread(udp1);
+ Thread udpThread1 = new Thread(server1);
udpThread1.start();
- Thread udpThread2 = new Thread(udp2);
+ Thread udpThread2 = new Thread(server2);
udpThread2.start();
try {