class Dish extends SocketBase
{
    // Fair queueing object for inbound pipes.
    private final FQ fq;

    // Object for distributing the subscriptions upstream.
    private final Dist dist;

    // The repository of subscriptions.
    private final Set<String> subscriptions;

    // If true, 'message' contains a matching message to return on the
    // next recv call.
    private Msg pendingMsg;

    //  Holds the prefetched message.
    public Dish(Ctx parent, int tid, int sid)
    {
        super(parent, tid, sid, true);

        options.type = ZMQ.ZMQ_DISH;

        // When socket is being closed down we don't want to wait till pending
        // subscription commands are sent to the wire.
        options.linger = 0;

        fq = new FQ();
        dist = new Dist();
        subscriptions = new HashSet<>();
    }

    @Override
    protected void xattachPipe(Pipe pipe, boolean subscribe2all, boolean isLocallyInitiated)
    {
        assert (pipe != null);

        fq.attach(pipe);
        dist.attach(pipe);

        // Send all the cached subscriptions to the new upstream peer.
        sendSubscriptions(pipe);
    }

    @Override
    protected void xreadActivated(Pipe pipe)
    {
        fq.activated(pipe);
    }

    @Override
    protected void xwriteActivated(Pipe pipe)
    {
        dist.activated(pipe);
    }

    @Override
    protected void xpipeTerminated(Pipe pipe)
    {
        fq.terminated(pipe);
        dist.terminated(pipe);
    }

    @Override
    protected void xhiccuped(Pipe pipe)
    {
        // Send all the cached subscriptions to the hiccuped pipe.
        sendSubscriptions(pipe);
    }

    @Override
    protected boolean xjoin(String group)
    {
        if (group.length() > Msg.MAX_GROUP_LENGTH) {
            errno.set(ZError.EINVAL);
            return false;
        }

        // User cannot join same group twice
        if (!subscriptions.add(group)) {
            errno.set(ZError.EINVAL);
            return false;
        }

        Msg msg = new Msg();
        msg.initJoin();
        msg.setGroup(group);

        dist.sendToAll(msg);

        return true;
    }

    @Override
    protected boolean xleave(String group)
    {
        if (group.length() > Msg.MAX_GROUP_LENGTH) {
            errno.set(ZError.EINVAL);
            return false;
        }

        if (!subscriptions.remove(group)) {
            errno.set(ZError.EINVAL);
            return false;
        }

        Msg msg = new Msg();
        msg.initLeave();
        msg.setGroup(group);

        dist.sendToAll(msg);

        return true;
    }

    @Override
    protected boolean xsend(Msg msg)
    {
        errno.set(ZError.ENOTSUP);

        throw new UnsupportedOperationException();
    }

    @Override
    protected Msg xrecv()
    {
        // If there's already a message prepared by a previous call to poll,
        // return it straight ahead.
        if (pendingMsg != null) {
            Msg msg = pendingMsg;
            pendingMsg = null;
            return msg;
        }

        return xxrecv();
    }

    private Msg xxrecv()
    {<FILL_FUNCTION_BODY>}

    @Override
    protected boolean xhasIn()
    {
        // If there's already a message prepared by a previous call to zmq_poll,
        // return straight ahead.
        if (pendingMsg != null) {
            return true;
        }

        Msg msg = xxrecv();
        if (msg == null) {
            return false;
        }

        //  Matching message found
        pendingMsg = msg;
        return true;
    }

    @Override
    protected boolean xhasOut()
    {
        // Subscription can be added/removed anytime.
        return true;
    }

    private void sendSubscriptions(Pipe pipe)
    {
        for (String s : subscriptions) {
            Msg msg = new Msg();
            msg.initJoin();
            msg.setGroup(s);
            pipe.write(msg);
        }

        pipe.flush();
    }

    public static class DishSession extends SessionBase
    {
        static final byte[] JOIN_BYTES = "\4JOIN".getBytes(StandardCharsets.US_ASCII);
        static final byte[] LEAVE_BYTES = "\5LEAVE".getBytes(StandardCharsets.US_ASCII);

        enum State
        {
            GROUP,
            BODY
        }

        private State state;
        private String group;

        public DishSession(IOThread ioThread, boolean connect, SocketBase socket, final Options options,
                           final Address addr)
        {
            super(ioThread, connect, socket, options, addr);

            state = State.GROUP;
            group = "";
        }

        @Override
        public boolean pushMsg(Msg msg)
        {
            switch (state) {
            case GROUP:
                if (!msg.hasMore()) {
                    errno.set(ZError.EFAULT);
                    return false;
                }

                if (msg.size() > Msg.MAX_GROUP_LENGTH) {
                    errno.set(ZError.EFAULT);
                    return false;
                }

                group = new String(msg.data(), StandardCharsets.US_ASCII);
                state = State.BODY;

                return true;
            case BODY:
                //  Set the message group
                msg.setGroup(group);

                //  Thread safe socket doesn't support multipart messages
                if (msg.hasMore()) {
                    errno.set(ZError.EFAULT);
                    return false;
                }

                //  Push message to dish socket
                boolean rc = super.pushMsg(msg);
                if (rc) {
                    state = State.GROUP;
                }

                return rc;
            default:
                throw new IllegalStateException();
            }
        }

        @Override
        protected Msg pullMsg()
        {
            Msg msg = super.pullMsg();
            if (msg == null) {
                return null;
            }

            if (!msg.isJoin() && !msg.isLeave()) {
                return msg;
            }

            Msg command;

            byte[] groupBytes = msg.getGroup().getBytes(StandardCharsets.US_ASCII);

            if (msg.isJoin()) {
                command = new Msg(groupBytes.length + 5);
                command.put(JOIN_BYTES);
            }
            else {
                command = new Msg(groupBytes.length + 6);
                command.put(LEAVE_BYTES);
            }

            command.setFlags(Msg.COMMAND);

            //  Copy the group
            command.put(groupBytes);

            return command;
        }

        @Override
        protected void reset()
        {
            super.reset();
            state = State.GROUP;
        }
    }
}

        // Get a message using fair queueing algorithm.
        Msg msg = fq.recv(errno);

        // If there's no message available, return immediately.
        // The same when error occurs.
        if (msg == null) {
            return null;
        }

        // Skip non matching messages
        while (!subscriptions.contains(msg.getGroup())) {
            msg = fq.recv(errno);
            if (msg == null) {
                return null;
            }
        }

        //  Found a matching message
        return msg;
    
<methods>public final boolean bind(java.lang.String) ,public final void cancel(java.util.concurrent.atomic.AtomicBoolean) ,public final void close() ,public final boolean connect(java.lang.String) ,public final int connectPeer(java.lang.String) ,public boolean disconnectPeer(int) ,public final int errno() ,public final void eventAcceptFailed(java.lang.String, int) ,public final void eventAccepted(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventBindFailed(java.lang.String, int) ,public final void eventCloseFailed(java.lang.String, int) ,public final void eventClosed(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventConnectDelayed(java.lang.String, int) ,public final void eventConnectRetried(java.lang.String, int) ,public final void eventConnected(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventDisconnected(java.lang.String, java.nio.channels.SelectableChannel) ,public final void eventHandshakeFailedAuth(java.lang.String, int) ,public final void eventHandshakeFailedNoDetail(java.lang.String, int) ,public final void eventHandshakeFailedProtocol(java.lang.String, int) ,public final void eventHandshakeSucceeded(java.lang.String, int) ,public final void eventHandshaken(java.lang.String, int) ,public final void eventListening(java.lang.String, java.nio.channels.SelectableChannel) ,public final java.nio.channels.SelectableChannel getFD() ,public final int getSocketOpt(int) ,public final java.lang.Object getSocketOptx(int) ,public final void hiccuped(zmq.pipe.Pipe) ,public final void inEvent() ,public final boolean join(java.lang.String) ,public final boolean leave(java.lang.String) ,public final boolean monitor(java.lang.String, int) ,public final void pipeTerminated(zmq.pipe.Pipe) ,public final int poll(int, int, java.util.concurrent.atomic.AtomicBoolean) ,public final void readActivated(zmq.pipe.Pipe) ,public final zmq.Msg recv(int) ,public final zmq.Msg recv(int, java.util.concurrent.atomic.AtomicBoolean) ,public final boolean send(zmq.Msg, int) ,public final boolean send(zmq.Msg, int, java.util.concurrent.atomic.AtomicBoolean) ,public final boolean setEventHook(zmq.ZMQ.EventConsummer, int) ,public final boolean setSocketOpt(int, java.lang.Object) ,public final boolean termEndpoint(java.lang.String) ,public java.lang.String toString() ,public java.lang.String typeString() ,public final void writeActivated(zmq.pipe.Pipe) <variables>private boolean active,protected java.lang.String connectRid,private final non-sealed java.util.concurrent.atomic.AtomicBoolean ctxTerminated,private final non-sealed java.util.concurrent.atomic.AtomicBoolean destroyed,private final non-sealed MultiMap<java.lang.String,zmq.SocketBase.EndpointPipe> endpoints,private java.nio.channels.SocketChannel fileDesc,private zmq.poll.Poller.Handle handle,private final non-sealed MultiMap<java.lang.String,zmq.pipe.Pipe> inprocs,private final non-sealed ThreadLocal<java.lang.Boolean> isInEventThreadLocal,private long lastTsc,private final non-sealed zmq.IMailbox mailbox,private final non-sealed AtomicReference<zmq.ZMQ.EventConsummer> monitor,private int monitorEvents,private final non-sealed Set<zmq.pipe.Pipe> pipes,private zmq.poll.Poller poller,private boolean rcvmore,private final non-sealed boolean threadSafe,private final non-sealed java.util.concurrent.locks.ReentrantLock threadSafeSync,private int ticks