diff options
Diffstat (limited to 'src/main/java/derms/net/rmulticast/ReliableMulticast.java')
| -rw-r--r-- | src/main/java/derms/net/rmulticast/ReliableMulticast.java | 14 |
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(); } |