TotalOrderMulticastReceiver

1 files changed, 53 insertions, 0 deletions

diff --git a/src/main/java/derms/net/tomulticast/Receive.java b/src/main/java/derms/net/tomulticast/Receive.java
new file mode 100644
index 0000000..f95ee5f
--- /dev/null
+++ b/src/main/java/derms/net/tomulticast/Receive.java
@@ -0,0 +1,53 @@
+package derms.net.tomulticast;
+
+import derms.net.rmulticast.MessagePayload;
+
+import java.io.IOException;
+import java.net.InetAddress;
+import java.net.InetSocketAddress;
+import java.util.concurrent.BlockingQueue;
+import java.util.concurrent.PriorityBlockingQueue;
+
+/**
+ * Receive messages from the multicast group and place them in the holdback queue.
+ * Once previous messages are received, they are transferred from the holdback queue
+ * to the delivery queue in total order.
+ */
+class Receive<T extends MessagePayload> extends TotalOrderMulticast<T> implements Runnable {
+    private final PriorityBlockingQueue<Message<T>> holdback;
+    private final BlockingQueue<Message<T>> deliver;
+
+    Receive(InetSocketAddress group, InetAddress laddr, BlockingQueue<Message<T>> deliver) throws IOException {
+        super(group, laddr);
+        this.holdback = new PriorityBlockingQueue<Message<T>>();
+        this.deliver = deliver;
+    }
+
+    @Override
+    public void run() {
+        try {
+            for (;;) {
+                Message<T> msg = sock.receive();
+                holdback.put(msg);
+                tryDeliver();
+            }
+        } catch (InterruptedException e) {
+            close();
+        }
+    }
+
+    // Try to move messages from the holdback queue to the delivery queue.
+    private void tryDeliver() {
+        while (!holdback.isEmpty()) {
+            // Messages with lower sequence numbers are removed from the priority queue first.
+            Message<T> msg = holdback.peek();
+            if (msg == null || msg.seq != seq+1)
+                return;
+
+            // This is the next element in the sequence; deliver it.
+            msg = holdback.remove();
+            deliver.add(msg);
+            incSeq();
+        }
+    }
+}