blob: bf583dd7d0620daccbd42e4f27b6802a9cfbba21 (
plain) (
blame)
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
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
|
package derms.net.rmulticast;
import derms.net.ConcurrentMulticastSocket;
import derms.net.Packet;
import java.io.IOException;
import java.net.DatagramPacket;
import java.net.InetAddress;
import java.net.InetSocketAddress;
import java.net.SocketTimeoutException;
import java.time.Duration;
import java.util.NoSuchElementException;
import java.util.Set;
import java.util.concurrent.BlockingQueue;
import java.util.logging.Logger;
class Receive<T extends MessagePayload> implements Runnable {
private static final int bufSize = 8192;
private static final Duration sockTimeout = Duration.ofMillis(500);
private final ConcurrentMulticastSocket inSock;
private final Set<MessageID> acks;
private final Set<MessageID> nacks; // Positively acknowledged messages.
private final ReceivedSet<T> received; // Negatively acknowledged messages.
private final BlockingQueue<Message<T>> retransmissions;
private final Set<InetAddress> groupMembers;
private final BlockingQueue<Message<T>> delivered;
private final Logger log;
Receive(InetSocketAddress group, Set<MessageID> acks, Set<MessageID> nacks, ReceivedSet<T> received, BlockingQueue<Message<T>> retransmissions, Set<InetAddress> groupMembers, BlockingQueue<Message<T>> delivered) throws IOException {
this.inSock = new ConcurrentMulticastSocket(group.getPort());
this.inSock.joinGroup(group.getAddress());
this.inSock.setSoTimeout(sockTimeout);
this.acks = acks;
this.nacks = nacks;
this.received = received;
this.retransmissions = retransmissions;
this.groupMembers = groupMembers;
this.delivered = delivered;
this.log = Logger.getLogger(this.getClass().getName());
}
@Override
public void run() {
log.info("Listening on " + inSock);
DatagramPacket pkt = new DatagramPacket(new byte[bufSize], bufSize);
for (;;) {
try {
inSock.receive(pkt);
Message<?> msg = Packet.decode(pkt, Message.class);
receive(msg);
log.info("Received " + msg);
} catch (SocketTimeoutException e) {
if (Thread.interrupted()) {
log.info("Interrupted. Shutting down.");
inSock.close();
return;
}
} catch (IOException | ClassNotFoundException | ClassCastException e) {
log.warning(e.getMessage());
}
}
}
private void receive(Message<?> msg) throws ClassCastException {
groupMembers.add(msg.sender);
receiveAcks(msg);
receiveNacks(msg);
if (msg instanceof HeartbeatMessage)
return; // Only deliver in-band messages.
deliver((Message<T>) msg);
}
private void receiveAcks(Message<?> msg) {
for (MessageID mid : msg.acks) {
acks.remove(mid);
if (!received.contains(mid))
nacks.add(mid);
}
}
private void receiveNacks(Message<?> msg) {
for (MessageID mid : msg.nacks) {
try {
Message<T> rmsg = received.getByID(mid);
retransmissions.add(rmsg);
} catch (NoSuchElementException e) {
nacks.add(mid);
}
}
}
private void deliver(Message<T> msg) {
acks.add(msg.id());
nacks.remove(msg.id());
received.add(msg);
retransmissions.remove(msg);
delivered.add(msg);
}
}
|