docs: def subscribe(
              self,
              subject="",
              queue="",
              cb=None,
              future=None,
              max_msgs=0,
              is_async=False,
              pending_msgs_limit=DEFAULT_SUB_PENDING_MSGS_LIMIT,
              pending_bytes_limit=DEFAULT_SUB_PENDING_BYTES_LIMIT,
      ):
          """
          Sends a SUB command to the server. Takes a queue parameter
          which can be used in case of distributed queues or left empty
          if it is not the case, and a callback that will be dispatched
          message for processing them.
          """
          if self.is_closed:
              raise ErrConnectionClosed
  
          if self.is_draining:
              raise ErrConnectionDraining
  
          self._ssid += 1
          sid = self._ssid
          sub = Subscription(
              subject=subject,
              queue=queue,
              cb=cb,
              future=future,
              max_msgs=max_msgs,
              is_async=is_async,
              sid=sid,
          )
          self._subs[sid] = sub
  
          if cb is not None:
              sub.pending_msgs_limit = pending_msgs_limit
              sub.pending_bytes_limit = pending_bytes_limit
              sub.pending_queue = tornado.queues.Queue(
                  maxsize=pending_msgs_limit)
  
              @tornado.gen.coroutine
              def wait_for_msgs():
                  while True:
                      sub = wait_for_msgs.sub
                      err_cb = wait_for_msgs.err_cb
  
                      try:
                          sub = wait_for_msgs.sub
                          if sub.closed:
                              break
  
                          msg = yield sub.pending_queue.get()
                          if msg is None:
                              break
  
                          sub.received += 1
                          sub.pending_size -= len(msg.data)
                          if sub.max_msgs > 0 and sub.received >= sub.max_msgs:
                              # If we have hit the max for delivered msgs, remove sub.
                              self._subs.pop(sub.sid, None)
                              self._remove_subscription(sub)
  
                          # Invoke depending of type of handler.
                          if sub.is_async:
                              # NOTE: Deprecate this usage in a next release,
                              # the handler implementation ought to decide
                              # the concurrency level at which the messages
                              # should be processed.
                              self._loop.spawn_callback(sub.cb, msg)
                          else:
                              yield sub.cb(msg)
                      except Exception as e:
                          # All errors from calling an async subscriber
                          # handler are async errors.
                          if err_cb is not None:
                              yield err_cb(e)
  
              # Bind the subscription and error cb if present
              wait_for_msgs.sub = sub
              wait_for_msgs.err_cb = self._error_cb
              self._loop.spawn_callback(wait_for_msgs)
  
          elif future is not None:
              # Used to handle the single response from a request
              # based on auto unsubscribe.
              sub.future = future
  
          # Send SUB command...
          sub_cmd = b''.join([
              SUB_OP, _SPC_,
              sub.subject.encode(), _SPC_,
              sub.queue.encode(), _SPC_, ("%d" % sid).encode(), _CRLF_
          ])
          yield self.send_command(sub_cmd)
          yield self._flush_pending()
          raise tornado.gen.Return(sid)