spqueue.java

package guide;

import java.util.ArrayList;

import org.zeromq.*;
import org.zeromq.ZMQ.Poller;
import org.zeromq.ZMQ.Socket;

//
// Simple Pirate queue
// This is identical to load-balancing  pattern, with no reliability mechanisms
// at all. It depends on the client for recovery. Runs forever.
//
public class spqueue
{

    private final static String WORKER_READY = "\001"; //  Signals worker is ready

    public static void main(String[] args)
    {
        try (ZContext ctx = new ZContext()) {
            Socket frontend = ctx.createSocket(SocketType.ROUTER);
            Socket backend = ctx.createSocket(SocketType.ROUTER);
            frontend.bind("tcp://*:5555"); //  For clients
            backend.bind("tcp://*:5556"); //  For workers

            //  Queue of available workers
            ArrayList<ZFrame> workers = new ArrayList<>();

            Poller poller = ctx.createPoller(2);
            poller.register(backend, Poller.POLLIN);
            poller.register(frontend, Poller.POLLIN);

            //  The body of this example is exactly the same as lruqueue2.
            while (true) {
                boolean workersAvailable = !workers.isEmpty();
                int rc = poller.poll(-1);

                //  Poll frontend only if we have available workers
                if (rc == -1)
                    break; //  Interrupted

                //  Handle worker activity on backend
                if (poller.pollin(0)) {
                    //  Use worker address for LRU routing
                    ZMsg msg = ZMsg.recvMsg(backend);
                    if (msg == null)
                        break; //  Interrupted
                    ZFrame address = msg.unwrap();
                    workers.add(address);

                    //  Forward message to client if it's not a READY
                    ZFrame frame = msg.getFirst();
                    if (new String(frame.getData(), ZMQ.CHARSET).equals(WORKER_READY))
                        msg.destroy();
                    else msg.send(frontend);
                }
                if (workersAvailable && poller.pollin(1)) {
                    //  Get client request, route to first available worker
                    ZMsg msg = ZMsg.recvMsg(frontend);
                    if (msg != null) {
                        msg.wrap(workers.remove(0));
                        msg.send(backend);
                    }
                }
            }

            //  When we're done, clean up properly
            while (!workers.isEmpty()) {
                ZFrame frame = workers.remove(0);
                frame.destroy();
            }

            workers.clear();
        }
    }
}