def request(self, subject, payload, timeout=0.5, expected=1, cb=None):
        """
        Implements the request/response pattern via pub/sub using an
        unique reply subject and an async subscription.

        If cb is None, then it will wait and return a single message
        using the new request/response style that is less chatty over
        the network.

          ->> SUB _INBOX.BF6zPVxvScfXGd4VyUMJyo.* 1
          ->> PUB hello _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo 5
          ->> MSG_PAYLOAD: hello
          <<- MSG hello 2 _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo 5
          ->> PUB _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo  5
          ->> MSG_PAYLOAD: world
          <<- MSG _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo 1 5

        If a cb is passed, then it will use auto unsubscribe
        functionality and expect a limited number of messages which
        will be handled asynchronously in the callback.

          ->> SUB _INBOX.gnKUg9bmAHANjxIsDiQsWO 90
          ->> UNSUB 90 1
          ->> PUB hello _INBOX.gnKUg9bmAHANjxIsDiQsWO 5
          ->> MSG_PAYLOAD: world
          <<- MSG hello 2 _INBOX.gnKUg9bmAHANjxIsDiQsWO 5

        """
        if self.is_draining:
            raise ErrConnectionDraining

        # If callback given then continue to use old style.
        if cb is not None:
            next_inbox = INBOX_PREFIX[:]
            next_inbox.extend(self._nuid.next())
            inbox = str(next_inbox)
            sid = yield self.subscribe(
                inbox,
                queue=_EMPTY_,
                cb=cb,
                max_msgs=expected,
            )
            yield self.auto_unsubscribe(sid, expected)
            yield self.publish_request(subject, inbox, payload)
            raise tornado.gen.Return(sid)

        if self._resp_sub_prefix is None:
            self._resp_map = {}

            # Create a prefix and single wildcard subscription once.
            self._resp_sub_prefix = str(INBOX_PREFIX[:])
            self._resp_sub_prefix += self._nuid.next()
            self._resp_sub_prefix += b'.'
            resp_mux_subject = str(self._resp_sub_prefix[:])
            resp_mux_subject += b'*'
            sub = Subscription(subject=str(resp_mux_subject))

            # FIXME: Allow setting pending limits for responses mux subscription.
            sub.pending_msgs_limit = DEFAULT_SUB_PENDING_MSGS_LIMIT
            sub.pending_bytes_limit = DEFAULT_SUB_PENDING_BYTES_LIMIT
            sub.pending_queue = tornado.queues.Queue(
                maxsize=sub.pending_msgs_limit)

            # Single task for handling the requests
            @tornado.gen.coroutine
            def wait_for_msgs():
                while True:
                    sub = wait_for_msgs.sub
                    if sub.closed:
                        break

                    msg = yield sub.pending_queue.get()
                    if msg is None:
                        break

                    token = msg.subject[INBOX_PREFIX_LEN:]
                    try:
                        fut = self._resp_map[token]
                        fut.set_result(msg)
                        del self._resp_map[token]
                    except KeyError:
                        # Future already handled so drop any extra
                        # responses which may have made it.
                        continue

            wait_for_msgs.sub = sub
            self._loop.spawn_callback(wait_for_msgs)

            # Store the subscription in the subscriptions map,
            # then send the protocol commands to the server.
            self._ssid += 1
            sid = self._ssid
            sub.sid = sid
            self._subs[sid] = sub

            # Send SUB command...
            sub_cmd = b''.join([
                SUB_OP, _SPC_,
                sub.subject.encode(), _SPC_, ("%d" % sid).encode(), _CRLF_
            ])
            yield self.send_command(sub_cmd)
            yield self._flush_pending()

        # Use a new NUID for the token inbox and then use the future.
        token = self._nuid.next()
        inbox = self._resp_sub_prefix[:]
        inbox.extend(token)
        future = tornado.concurrent.Future()
        self._resp_map[token.decode()] = future
        yield self.publish_request(subject, str(inbox), payload)
        msg = yield tornado.gen.with_timeout(
            timedelta(seconds=timeout), future)
        raise tornado.gen.Return(msg)
def request ( self , subject , payload , timeout = 0.5 , expected = 1 , cb = None ) : if self . is_draining : raise ErrConnectionDraining if cb is not None : next_inbox = INBOX_PREFIX [ : ] next_inbox . extend ( self . _nuid . next ( ) ) inbox = str ( next_inbox ) sid = yield self . subscribe ( inbox , queue = _EMPTY_ , cb = cb , max_msgs = expected , ) yield self . auto_unsubscribe ( sid , expected ) yield self . publish_request ( subject , inbox , payload ) raise tornado . gen . Return ( sid ) if self . _resp_sub_prefix is None : self . _resp_map = { } self . _resp_sub_prefix = str ( INBOX_PREFIX [ : ] ) self . _resp_sub_prefix += self . _nuid . next ( ) self . _resp_sub_prefix += b'.' resp_mux_subject = str ( self . _resp_sub_prefix [ : ] ) resp_mux_subject += b'*' sub = Subscription ( subject = str ( resp_mux_subject ) ) sub . pending_msgs_limit = DEFAULT_SUB_PENDING_MSGS_LIMIT sub . pending_bytes_limit = DEFAULT_SUB_PENDING_BYTES_LIMIT sub . pending_queue = tornado . queues . Queue ( maxsize = sub . pending_msgs_limit ) @ tornado . gen . coroutine def wait_for_msgs ( ) : while True : sub = wait_for_msgs . sub if sub . closed : break msg = yield sub . pending_queue . get ( ) if msg is None : break token = msg . subject [ INBOX_PREFIX_LEN : ] try : fut = self . _resp_map [ token ] fut . set_result ( msg ) del self . _resp_map [ token ] except KeyError : continue wait_for_msgs . sub = sub self . _loop . spawn_callback ( wait_for_msgs ) self . _ssid += 1 sid = self . _ssid sub . sid = sid self . _subs [ sid ] = sub sub_cmd = b'' . join ( [ SUB_OP , _SPC_ , sub . subject . encode ( ) , _SPC_ , ( "%d" % sid ) . encode ( ) , _CRLF_ ] ) yield self . send_command ( sub_cmd ) yield self . _flush_pending ( ) token = self . _nuid . next ( ) inbox = self . _resp_sub_prefix [ : ] inbox . extend ( token ) future = tornado . concurrent . Future ( ) self . _resp_map [ token . decode ( ) ] = future yield self . publish_request ( subject , str ( inbox ) , payload ) msg = yield tornado . gen . with_timeout ( timedelta ( seconds = timeout ) , future ) raise tornado . gen . Return ( msg )
Implements the request/response pattern via pub/sub using an
        unique reply subject and an async subscription.

        If cb is None, then it will wait and return a single message
        using the new request/response style that is less chatty over
        the network.

          ->> SUB _INBOX.BF6zPVxvScfXGd4VyUMJyo.* 1
          ->> PUB hello _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo 5
          ->> MSG_PAYLOAD: hello
          <<- MSG hello 2 _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo 5
          ->> PUB _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo  5
          ->> MSG_PAYLOAD: world
          <<- MSG _INBOX.BF6zPVxvScfXGd4VyUMJyo.BF6zPVxvScfXCh4VyUMJyo 1 5

        If a cb is passed, then it will use auto unsubscribe
        functionality and expect a limited number of messages which
        will be handled asynchronously in the callback.

          ->> SUB _INBOX.gnKUg9bmAHANjxIsDiQsWO 90
          ->> UNSUB 90 1
          ->> PUB hello _INBOX.gnKUg9bmAHANjxIsDiQsWO 5
          ->> MSG_PAYLOAD: world
          <<- MSG hello 2 _INBOX.gnKUg9bmAHANjxIsDiQsWO 5