@Override
    protected Msg xrecv (int flags_) {
        //  If there's already a message prepared by a previous call to zmq_poll,
        //  return it straight ahead.
        Msg msg_;
        if (has_message) {
            msg_ = new Msg(message);
            has_message = false;
            more = msg_.has_more();
            return msg_;
        }

        //  TODO: This can result in infinite loop in the case of continuous
        //  stream of non-matching messages which breaks the non-blocking recv
        //  semantics.
        while (true) {

            //  Get a message using fair queueing algorithm.
            msg_ = fq.recv ();

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

            //  Check whether the message matches at least one subscription.
            //  Non-initial parts of the message are passed 
            if (more || !options.filter || match (msg_)) {
                more = msg_.has_more();
                return msg_;
            }

            //  Message doesn't match. Pop any remaining parts of the message
            //  from the pipe.
            while (msg_.has_more()) {
                msg_ = fq.recv ();
                assert (msg_ != null);
            }
        }
    }