class Outpipe
    {
        private final Pipe    pipe;
        private boolean active;

        public Outpipe(Pipe pipe, boolean active)
        {
            this.pipe = pipe;
            this.active = active;
        }
    }

    //  We keep a set of pipes that have not been identified yet.
    private final Set<Pipe> anonymousPipes;

    //  Outbound pipes indexed by the peer IDs.
    private final Map<Blob, Outpipe> outpipes;

    //  The pipe we are currently writing to.
    private Pipe currentOut;

    //  If true, more outgoing message parts are expected.
    private boolean moreOut;

    //  Routing IDs are generated. It's a simple increment and wrap-over
    //  algorithm. This value is the next ID to use (if not used already).
    private int nextRid;

    // If true, report EAGAIN to the caller instead of silently dropping
    // the message targeting an unknown peer.
    private boolean mandatory;

    private boolean rawSocket;

    // if true, send an empty message to every connected router peer
    private boolean probeRouter;

    // If true, the router will reassign an identity upon encountering a
    // name collision. The new pipe will take the identity, the old pipe
    // will be terminated.
    private boolean handover;

    public Router(Ctx parent, int tid, int sid)
    {
        super(parent, tid, sid);
        prefetched = false;
        identitySent = false;
        moreIn = false;
        currentOut = null;
        moreOut = false;
        nextRid = Utils.randomInt();
        mandatory = false;
        //  raw_sock functionality in ROUTER is deprecated
        rawSocket = false;
        handover = false;

        options.type = ZMQ.ZMQ_ROUTER;
        options.recvIdentity = true;
        options.rawSocket = false;
        options.canSendHelloMsg = true;
        options.canReceiveDisconnectMsg = true;

        fq = new FQ();
        prefetchedId = new Msg();
        prefetchedMsg = new Msg();

        anonymousPipes = new HashSet<>();
        outpipes = new HashMap<>();
    }

    @Override
    protected void destroy()
    {
        assert (anonymousPipes.isEmpty());
        assert (outpipes.isEmpty());

        super.destroy();
    }

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

        if (probeRouter) {
            Msg probe = new Msg();
            pipe.write(probe);
            // assert (rc) is not applicable here, since it is not a bug.
            pipe.flush();
        }
        boolean identityOk = identifyPeer(pipe, isLocallyInitiated);
        if (identityOk) {
            fq.attach(pipe);
        }
        else {
            anonymousPipes.add(pipe);
        }
    }

    @Override
    public boolean xsetsockopt(int option, Object optval)
    {
        if (option == ZMQ.ZMQ_CONNECT_RID) {
            connectRid = Options.parseString(option, optval);
            return true;
        }
        if (option == ZMQ.ZMQ_ROUTER_RAW) {
            rawSocket = Options.parseBoolean(option, optval);
            if (rawSocket) {
                options.recvIdentity = false;
                options.rawSocket = true;
            }
            return true;
        }
        if (option == ZMQ.ZMQ_ROUTER_MANDATORY) {
            mandatory = Options.parseBoolean(option, optval);
            return true;
        }
        if (option == ZMQ.ZMQ_PROBE_ROUTER) {
            probeRouter = Options.parseBoolean(option, optval);
            return true;
        }
        if (option == ZMQ.ZMQ_ROUTER_HANDOVER) {
            handover = Options.parseBoolean(option, optval);
            return true;
        }
        errno.set(ZError.EINVAL);
        return false;
    }

    @Override
    public void xpipeTerminated(Pipe pipe)
    {
        if (!anonymousPipes.remove(pipe)) {
            Outpipe old = outpipes.remove(pipe.getIdentity());
            assert (old != null);

            fq.terminated(pipe);
            if (pipe == currentOut) {
                currentOut = null;
            }
        }
    }

    @Override
    public void xreadActivated(Pipe pipe)
    {
        if (!anonymousPipes.contains(pipe)) {
            fq.activated(pipe);
        }
        else {
            boolean identityOk = identifyPeer(pipe, false);
            if (identityOk) {
                anonymousPipes.remove(pipe);
                fq.attach(pipe);
            }
        }
    }

    @Override
    public void xwriteActivated(Pipe pipe)
    {
        for (Outpipe out : outpipes.values()) {
            if (out.pipe == pipe) {
                assert (!out.active);
                out.active = true;
                break;
            }
        }
    }

    @Override
    protected boolean xsend(Msg msg)
    {<FILL_FUNCTION_BODY>

        //  If this is the first part of the message it's the ID of the
        //  peer to send the message to.
        if (!moreOut) {
            assert (currentOut == null);

            //  If we have malformed message (prefix with no subsequent message)
            //  then just silently ignore it.
            //  TODO: The connections should be killed instead.
            if (msg.hasMore()) {
                moreOut = true;

                //  Find the pipe associated with the identity stored in the prefix.
                //  If there's no such pipe just silently ignore the message, unless
                //  mandatory is set.
                Blob identity = Blob.createBlob(msg);
                Outpipe op = outpipes.get(identity);

                if (op != null) {
                    currentOut = op.pipe;
                    if (!currentOut.checkWrite()) {
                        op.active = false;
                        currentOut = null;
                        if (mandatory) {
                            moreOut = false;
                            errno.set(ZError.EAGAIN);
                            return false;
                        }
                    }
                }
                else if (mandatory) {
                    moreOut = false;
                    errno.set(ZError.EHOSTUNREACH);
                    return false;
                }
            }

            return true;
        }

        //  Ignore the MORE flag for raw-sock or assert?
        if (options.rawSocket) {
            msg.resetFlags(Msg.MORE);
        }

        //  Check whether this is the last part of the message.
        moreOut = msg.hasMore();

        //  Push the message into the pipe. If there's no out pipe, just drop it.
        if (currentOut != null) {
            // Close the remote connection if user has asked to do so
            // by sending zero length message.
            // Pending messages in the pipe will be dropped (on receiving term- ack)
            if (rawSocket && msg.size() == 0) {
                currentOut.terminate(false);
                currentOut = null;
                return true;
            }

            boolean ok = currentOut.write(msg);
            if (!ok) {
                // Message failed to send - we must close it ourselves.
                currentOut = null;
            }
            else if (!moreOut) {
                currentOut.flush();
                currentOut = null;
            }
        }

        return true;
    
<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