summaryrefslogtreecommitdiffstats
path: root/src/main/java/derms/net/rmulticast/Receive.java
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java/derms/net/rmulticast/Receive.java')
-rw-r--r--src/main/java/derms/net/rmulticast/Receive.java5
1 files changed, 4 insertions, 1 deletions
diff --git a/src/main/java/derms/net/rmulticast/Receive.java b/src/main/java/derms/net/rmulticast/Receive.java
index c8467c2..c1965ed 100644
--- a/src/main/java/derms/net/rmulticast/Receive.java
+++ b/src/main/java/derms/net/rmulticast/Receive.java
@@ -18,14 +18,16 @@ class Receive<T extends Serializable & Hashable> implements Runnable {
private final Set<MessageID> negativeAcks;
private final ReceivedSet<T> received;
private final BlockingQueue<Message<T>> retransmissions;
+ private final BlockingQueue<Message<T>> delivered;
private final Logger log;
- Receive(ConcurrentMulticastSocket inSock, Set<MessageID> positiveAcks, Set<MessageID> negativeAcks, ReceivedSet<T> received, BlockingQueue<Message<T>> retransmissions) {
+ Receive(ConcurrentMulticastSocket inSock, Set<MessageID> positiveAcks, Set<MessageID> negativeAcks, ReceivedSet<T> received, BlockingQueue<Message<T>> retransmissions, BlockingQueue<Message<T>> delivered) {
this.inSock = inSock;
this.positiveAcks = positiveAcks;
this.negativeAcks = negativeAcks;
this.received = received;
this.retransmissions = retransmissions;
+ this.delivered = delivered;
this.log = Logger.getLogger(this.getClass().getName());
}
@@ -47,6 +49,7 @@ class Receive<T extends Serializable & Hashable> implements Runnable {
private void receive(Message<T> msg) {
positiveAcks.add(msg.id());
received.add(msg);
+ delivered.add(msg);
negativeAcks.remove(msg.id());
retransmissions.remove(msg);