package bench.pubsub;

import org.zeromq.ZMQ;

/**
 * This sequencer simply reads from a PULL socket and forwards the
 * message to the PUB socket.
 */

public class ZeroMQSequencer {
    
    public static void main (String[] args) throws Exception {
        new Thread(new Sequencer()).start();
    }

    public static class Sequencer implements Runnable {
        public void run() {
            //  Prepare our context and publisher
            ZMQ.Context context = ZMQ.context(2);
            

            ZMQ.Socket publisher = context.socket(ZMQ.PUB);
            publisher.setAffinity(1);
            //publisher.setSndHWM(100000);
            publisher.bind("tcp://*:5556");
            //publisher.bind("ipc://weather");

            ZMQ.Socket incoming = context.socket(ZMQ.PULL);
            //incoming.setRcvHWM(100000);
            incoming.setAffinity(2);
            incoming.bind("tcp://*:5557");

            ZMQ.proxy(incoming, publisher, null);
            // while (! Thread.currentThread().isInterrupted()) {
            //     byte[] data = incoming.recv(0);
            //     //System.out.printf("Received message with %d bytes\n", data.length);
            //     publisher.send(data, 0);
            // }

            publisher.close();
            incoming.close();
            context.term();
        }
    }
}

@@ -177,20 +177,16 @@
 encer %7B%0A
-    
 %0A    pub
@@ -205,17 +205,16 @@
 oid main
- 
 (String%5B
@@ -348,16 +348,34 @@
 nable %7B%0A
+        @Override%0A
         
@@ -493,29 +493,16 @@
 text(2);
-%0A            
 %0A%0A      
@@ -597,34 +597,32 @@
 1);%0A            
-//
 publisher.setSnd
@@ -622,20 +622,12 @@
 .set
-Snd
 HWM(
-10000
 0);%0A
@@ -790,18 +790,16 @@
         
-//
 incoming
@@ -806,20 +806,12 @@
 .set
-Rcv
 HWM(
-10000
 0);%0A
@@ -887,24 +887,79 @@
 /*:5557%22);%0A%0A
+            System.out.println(%22Sequencer is alive%22);%0A%0A
             

Set Sequencer's HighWaterMark to 0