refactor ReliableMulticast.Retransmit to top-level class

2 files changed, 45 insertions, 29 deletions

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<T extends Serializable & Hashable> {
         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<T extends Serializable & Hashable> {
         }
     }
 
-    /** 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<T> 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<T extends Serializable & Hashable> implements Runnable {
+    private final BlockingQueue<Message<T>> retransmissions;
+    private final MulticastSocket outSock;
+    private final SocketAddress group;
+    private final Logger log;
+
+    Retransmit(BlockingQueue<Message<T>> 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<T> 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());
+        }
+    }
+}