1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
|
package pl.edu.mimuw.cloudatlas.agent.modules;
import com.google.common.primitives.Bytes;
import pl.edu.mimuw.cloudatlas.agent.messages.UDUPMessage;
import java.io.IOException;
import java.net.*;
import java.util.HashMap;
public class UDUPServer {
private UDUP udp;
private UDUPSerializer serializer;
private DatagramSocket socket;
private InetAddress address;
private HashMap<InetAddress, byte[]> partialPackets;
private int bufSize;
public UDUPServer(UDUP udp, InetAddress addr, int port, int bufSize) throws SocketException {
this.udp = udp;
this.socket = new DatagramSocket(port, addr);
this.address = addr;
this.bufSize = bufSize;
this.partialPackets = new HashMap<>();
this.serializer = new UDUPSerializer();
}
public void acceptMessage() throws IOException, InterruptedException {
byte[] buf = new byte[bufSize];
DatagramPacket packet = new DatagramPacket(buf, buf.length);
this.socket.receive(packet);
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 {
if (msg.getContent().getDestinationModule() == ModuleType.TEST) {
System.out.println("UDP server: test message received");
} else if (msg.getContent().getDestinationModule() != ModuleType.UDP) {
this.udp.sendMessage(msg.getContent());
}
}
} else {
this.addPartialMessageAndCheckSerialization(packet.getAddress(), packet.getData());
}
}
public void addPartialMessageAndCheckSerialization(InetAddress senderAddress, byte[] packetData) {
if (this.partialPackets.containsKey(senderAddress)) {
byte[] previousPacketData = this.partialPackets.get(senderAddress);
byte[] allPacketData = Bytes.concat(previousPacketData, packetData);
try {
UDUPMessage msg = this.serializer.deserialize(allPacketData);
this.udp.sendMessage(msg.getContent());
this.partialPackets.remove(senderAddress);
} catch (Error | Exception e) {
System.out.println("Kryo didn't deserialize partial message, waiting to receive the rest");
this.partialPackets.put(senderAddress, allPacketData);
}
} else {
this.partialPackets.put(senderAddress, packetData);
}
}
public void close() {
this.socket.close();
}
}
|