class MarkAsMatching implements IMtrieHandler
    {
        @Override
        public void invoke(Pipe pipe, byte[] data, int size, XPub self)
        {
            self.markAsMatching(pipe);
        }
    }

    //  List of all subscriptions mapped to corresponding pipes.
    private final Mtrie subscriptions;

    //  List of manual subscriptions mapped to corresponding pipes.
    private final Mtrie manualSubscriptions;

    //  Distributor of messages holding the list of outbound pipes.
    private final Dist dist;

    // If true, send all subscription messages upstream, not just
    // unique ones
    private boolean verboseSubs;

    // If true, send all unsubscription messages upstream, not just
    // unique ones
    private boolean verboseUnsubs;

    //  True if we are in the middle of sending a multi-part message.
    private boolean more;

    //  Drop messages if HWM reached, otherwise return with false
    private boolean lossy;

    //  Subscriptions will not bed added automatically, only after calling set option with ZMQ_SUBSCRIBE or ZMQ_UNSUBSCRIBE
    private boolean manual;

    //  Last pipe that sent subscription message, only used if xpub is on manual
    private Pipe lastPipe;

    // Pipes that sent subscriptions messages that have not yet been processed, only used if xpub is on manual
    private final Deque<Pipe> pendingPipes;

    //  List of pending (un)subscriptions, ie. those that were already
    //  applied to the trie, but not yet received by the user.
    private final Deque<Blob>    pendingData;
    private final Deque<Integer> pendingFlags;

    private static final IMtrieHandler markAsMatching     = new MarkAsMatching();
    private static final IMtrieHandler sendUnsubscription = new SendUnsubscription();

    public XPub(Ctx parent, int tid, int sid)
    {
        super(parent, tid, sid);

        options.type = ZMQ.ZMQ_XPUB;
        verboseSubs = false;
        verboseUnsubs = false;
        more = false;
        lossy = true;
        manual = false;

        subscriptions = new Mtrie();
        manualSubscriptions = new Mtrie();
        dist = new Dist();
        lastPipe = null;
        pendingPipes = new ArrayDeque<>();
        pendingData = new ArrayDeque<>();
        pendingFlags = new ArrayDeque<>();
    }

    @Override
    protected void xattachPipe(Pipe pipe, boolean subscribeToAll, boolean isLocallyInitiated)
    {
        assert (pipe != null);
        dist.attach(pipe);

        //  If subscribe_to_all_ is specified, the caller would like to subscribe
        //  to all data on this pipe, implicitly.
        if (subscribeToAll) {
            subscriptions.addOnTop(pipe);
        }

        //  The pipe is active when attached. Let's read the subscriptions from
        //  it, if any.
        xreadActivated(pipe);
    }

    @Override
    protected void xreadActivated(Pipe pipe)
    {<FILL_FUNCTION_BODY>

        //  There are some subscriptions waiting. Let's process them.
        Msg sub;
        while ((sub = pipe.read()) != null) {
            //  Apply the subscription to the trie
            boolean subscribe;
            int size = sub.size();
            if (size > 0 && (sub.get(0) == 0 || sub.get(0) == 1)) {
                subscribe = sub.get(0) == 1;
            }
            else {
                //  Process user message coming upstream from xsub socket
                pendingData.add(Blob.createBlob(sub));
                pendingFlags.add(sub.flags());
                continue;
            }

            if (manual) {
                // Store manual subscription to use on termination
                if (!subscribe) {
                    manualSubscriptions.rm(sub, pipe);
                }
                else {
                    manualSubscriptions.add(sub, pipe);
                }

                pendingPipes.add(pipe);

                //  ZMTP 3.1 hack: we need to support sub/cancel commands, but
                //  we can't give them back to userspace as it would be an API
                //  breakage since the payload of the message is completely
                //  different. Manually craft an old-style message instead.
                pendingData.add(Blob.createBlob(sub));
                pendingFlags.add(0);
            }
            else {
                boolean notify;
                if (!subscribe) {
                    notify = subscriptions.rm(sub, pipe) || verboseUnsubs;
                }
                else {
                    notify = subscriptions.add(sub, pipe) || verboseSubs;
                }

                //  If the request was a new subscription, or the subscription
                //  was removed, or verbose mode is enabled, store it so that
                //  it can be passed to the user on next recv call.
                if (options.type == ZMQ.ZMQ_XPUB && notify) {
                    pendingData.add(Blob.createBlob(sub));
                    pendingFlags.add(0);
                }
            }
        }
    
<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