From 24e443366e31b194acac8327cb2b7fdb6a2cca89 Mon Sep 17 00:00:00 2001 From: Sam Anthony Date: Thu, 14 Nov 2024 11:53:43 -0500 Subject: refactor ReliableMulticast.Retransmit to top-level class --- .../derms/net/rmulticast/ReliableMulticast.java | 30 +-------------- src/main/java/derms/net/rmulticast/Retransmit.java | 44 ++++++++++++++++++++++ 2 files changed, 45 insertions(+), 29 deletions(-) create mode 100644 src/main/java/derms/net/rmulticast/Retransmit.java (limited to 'src/main/java/derms') diff --git a/src/main/java/derms/net/rmulticast/ReliableMulticast.java b/src/main/java/derms/net/rmulticast/ReliableMulticast.java index f20dd2a..f63433f 100644 --- a/src/main/java/derms/net/rmulticast/ReliableMulticast.java +++ b/src/main/java/derms/net/rmulticast/ReliableMulticast.java @@ -48,7 +48,7 @@ public class ReliableMulticast { this.log = Logger.getLogger(this.getClass().getName()); (new Thread(new Receive())).start(); - (new Thread(new Retransmit())).start(); + (new Thread(new Retransmit(retransmissions, outSock, group))).start(); } public void send(T payload) throws IOException { @@ -73,34 +73,6 @@ public class ReliableMulticast { } } - /** Retransmit dropped messages. */ - private class Retransmit implements Runnable { - private final Logger log; - - private Retransmit() { - this.log = Logger.getLogger(this.getClass().getName()); - } - - @Override - public void run() { - try { - for (;;) { - Message msg = retransmissions.take(); - try { - DatagramPacket pkt = Packet.encode(msg, group); - synchronized (outSock) { - outSock.send(pkt); - } - } catch (Exception e) { - log.warning(e.getMessage()); - } - } - } catch (InterruptedException e) { - log.info("Retransmit thread interrupted: "+e.getMessage()); - } - } - } - private class Timeout implements Runnable { MessageID mid; diff --git a/src/main/java/derms/net/rmulticast/Retransmit.java b/src/main/java/derms/net/rmulticast/Retransmit.java new file mode 100644 index 0000000..ad44eb1 --- /dev/null +++ b/src/main/java/derms/net/rmulticast/Retransmit.java @@ -0,0 +1,44 @@ +package derms.net.rmulticast; + +import derms.net.Packet; + +import java.io.Serializable; +import java.net.DatagramPacket; +import java.net.MulticastSocket; +import java.net.SocketAddress; +import java.util.concurrent.BlockingQueue; +import java.util.logging.Logger; + +/** Retransmit dropped messages. */ +class Retransmit implements Runnable { + private final BlockingQueue> retransmissions; + private final MulticastSocket outSock; + private final SocketAddress group; + private final Logger log; + + Retransmit(BlockingQueue> retransmissions, MulticastSocket outSock, SocketAddress group) { + this.retransmissions = retransmissions; + this.outSock = outSock; + this.group = group; + this.log = Logger.getLogger(this.getClass().getName()); + } + + @Override + public void run() { + try { + for (; ; ) { + Message msg = retransmissions.take(); + try { + DatagramPacket pkt = Packet.encode(msg, group); + synchronized (outSock) { + outSock.send(pkt); + } + } catch (Exception e) { + log.warning(e.getMessage()); + } + } + } catch (InterruptedException e) { + log.info("Retransmit thread interrupted: " + e.getMessage()); + } + } +} -- cgit v1.2.3