@Override
    protected Msg xrecv (int flags_)
    {
        Msg msg_ = null;
        if (prefetched) {
            if (!identity_sent) {
                msg_ = prefetched_id;
                prefetched_id = null;
                identity_sent = true;
            }
            else {
                msg_ = prefetched_msg;
                prefetched_msg = null;
                prefetched = false;
            }
            more_in = msg_.has_more();
            return msg_;
        }

        Pipe[] pipe = new Pipe[1];
        msg_ = fq.recvpipe (pipe);
        
        //  It's possible that we receive peer's identity. That happens
        //  after reconnection. The current implementation assumes that
        //  the peer always uses the same identity.
        //  TODO: handle the situation when the peer changes its identity.
        while (msg_ != null && msg_.is_identity ())
            msg_ = fq.recvpipe (pipe);

        if (msg_ == null)
            return null;
        
        assert (pipe[0] != null);

        //  If we are in the middle of reading a message, just return the next part.
        if (more_in)
            more_in = msg_.has_more();
        else {
            //  We are at the beginning of a message.
            //  Keep the message part we have in the prefetch buffer
            //  and return the ID of the peer instead.
            prefetched_msg = msg_;

            prefetched = true;

            Blob identity = pipe[0].get_identity ();
            msg_ = new Msg(identity.data());
            msg_.set_flags (Msg.more);
            identity_sent = true;
        }

        return msg_;
    }