@Override
    protected void xread_activated (Pipe pipe_)
    {
        //  There are some subscriptions waiting. Let's process them.
        Msg sub = null;
        while ((sub = pipe_.read()) != null) {

            //  Apply the subscription to the trie.
            byte[] data = sub.data();
            int size = sub.size ();
            if (size > 0 && (data[0] == 0 || data[0] == 1)) {
                boolean unique;
                if (data[0] == 0)
                    unique = subscriptions.rm (data , 1, pipe_);
                else
                    unique = subscriptions.add (data , 1, pipe_);

                //  If the subscription is not a duplicate, store it so that it can be
                //  passed to used on next recv call. (Unsubscribe is not verbose.)
                if (options.type == ZMQ.ZMQ_XPUB && (unique || (data[0] == 1 && verbose)))
                    pending.add (new Blob (sub.data ()));
            }
        }
    }