rmulticast: use DatagramChannel

1 files changed, 18 insertions, 17 deletions

diff --git a/src/main/java/derms/net/rmulticast/Retransmit.java b/src/main/java/derms/net/rmulticast/Retransmit.java
index 9d5e10b..0d2a604 100644
--- a/src/main/java/derms/net/rmulticast/Retransmit.java
+++ b/src/main/java/derms/net/rmulticast/Retransmit.java
@@ -1,43 +1,44 @@
 package derms.net.rmulticast;
 
-import derms.net.ConcurrentMulticastSocket;
+import derms.io.Serial;
 import derms.net.MessagePayload;
-import derms.net.Packet;
 
-import java.net.DatagramPacket;
+import java.io.IOException;
 import java.net.SocketAddress;
+import java.nio.ByteBuffer;
+import java.nio.channels.ClosedChannelException;
+import java.nio.channels.DatagramChannel;
 import java.util.concurrent.BlockingQueue;
 import java.util.logging.Logger;
 
 /** Retransmit dropped messages. */
 class Retransmit<T extends MessagePayload> implements Runnable {
     private final BlockingQueue<Message<T>> retransmissions;
-    private final ConcurrentMulticastSocket outSock;
+    private final DatagramChannel sock;
     private final SocketAddress group;
     private final Logger log;
 
-    Retransmit(BlockingQueue<Message<T>> retransmissions, ConcurrentMulticastSocket outSock, SocketAddress group) {
+    Retransmit(BlockingQueue<Message<T>> retransmissions, DatagramChannel sock, SocketAddress group) {
         this.retransmissions = retransmissions;
-        this.outSock = outSock;
+        this.sock = sock;
         this.group = group;
         this.log = Logger.getLogger(this.getClass().getName());
     }
 
     @Override
     public void run() {
-        try {
-            for (; ; ) {
+        for (;;) {
+            try {
                 Message<T> msg = retransmissions.take();
-                try {
-                    DatagramPacket pkt = Packet.encode(msg, group);
-                    outSock.send(pkt);
-                    log.info("Retransmitted " + msg);
-                } catch (Exception e) {
-                    log.warning(e.getMessage());
-                }
+                ByteBuffer buf = Serial.encode(msg);
+                sock.send(buf, group);
+                log.info("Retransmitted " + msg);
+            } catch (InterruptedException | ClosedChannelException e) {
+                log.info("Shutting down.");
+                return;
+            } catch (IOException e) {
+                log.warning(e.getMessage());
             }
-        } catch (InterruptedException e) {
-            log.info("Interrupted. Shutting down.");
         }
     }
 }