summaryrefslogtreecommitdiffstats
path: root/src/main/java/derms/net/rmulticast/ReliableMulticast.java
diff options
context:
space:
mode:
Diffstat (limited to 'src/main/java/derms/net/rmulticast/ReliableMulticast.java')
-rw-r--r--src/main/java/derms/net/rmulticast/ReliableMulticast.java14
1 files changed, 8 insertions, 6 deletions
diff --git a/src/main/java/derms/net/rmulticast/ReliableMulticast.java b/src/main/java/derms/net/rmulticast/ReliableMulticast.java
index 8b39ae6..de07f39 100644
--- a/src/main/java/derms/net/rmulticast/ReliableMulticast.java
+++ b/src/main/java/derms/net/rmulticast/ReliableMulticast.java
@@ -16,28 +16,30 @@ import java.util.logging.Logger;
/** TODO */
public class ReliableMulticast<T extends Serializable & Hashable> {
+ private final SocketAddress group;
+ private final InetAddress laddr; // Local address.
private final Set<MessageID> positiveAcks; // Positively acknowledged messages.
private final Set<MessageID> negativeAcks; // Negatively acknowledged messages.
private final BlockingQueue<Message<T>> retransmissions; // Messages pending retransmission.
private final AtomicReference<Instant> lastSend;
- private final SocketAddress group;
private final ConcurrentMulticastSocket outSock;
- private final InetAddress laddr; // Local address.
+ private final Set<InetAddress> groupMembers;
private final BlockingQueue<Message<T>> delivered;
private final Logger log;
public ReliableMulticast(InetSocketAddress group, InetAddress laddr) throws IOException {
+ this.group = group;
+ this.laddr = laddr;
+
this.positiveAcks = new ConcurrentHashMap<MessageID, Void>().keySet();
this.negativeAcks = new ConcurrentHashMap<MessageID, Void>().keySet();
this.retransmissions = new LinkedBlockingQueue<Message<T>>();
this.lastSend = new AtomicReference<Instant>(Instant.now());
- this.group = group;
-
this.outSock = new ConcurrentMulticastSocket(group.getPort());
this.outSock.joinGroup(group.getAddress());
- this.laddr = laddr;
+ this.groupMembers = new ConcurrentHashMap<InetAddress, Void>().keySet();
this.delivered = new LinkedBlockingQueue<Message<T>>();
@@ -45,7 +47,7 @@ public class ReliableMulticast<T extends Serializable & Hashable> {
ConcurrentMulticastSocket inSock = new ConcurrentMulticastSocket();
inSock.joinGroup(group.getAddress());
- (new Thread(new Receive<T>(inSock, positiveAcks, negativeAcks, retransmissions, delivered))).start();
+ (new Thread(new Receive<T>(inSock, positiveAcks, negativeAcks, retransmissions, groupMembers, delivered))).start();
(new Thread(new Retransmit<T>(retransmissions, outSock, group))).start();
}