docs: 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)