async def subscribe(self, subject,<EOL>cb=None,<EOL>error_cb=None,<EOL>start_at=None,<EOL>deliver_all_available=False,<EOL>sequence=None,<EOL>time=None,<EOL>manual_acks=False,<EOL>queue=None,<EOL>ack_wait=DEFAULT_ACK_WAIT,<EOL>max_inflight=DEFAULT_MAX_INFLIGHT,<EOL>durable_name=None,<EOL>pending_limits=DEFAULT_PENDING_LIMIT,<EOL>):
sub = Subscription(<EOL>subject=subject,<EOL>queue=queue,<EOL>cb=cb,<EOL>error_cb=error_cb,<EOL>manual_acks=manual_acks,<EOL>stan=self,<EOL>)<EOL>self._sub_map[sub.inbox] = sub<EOL>sub._msgs_queue = asyncio.Queue(maxsize=pending_limits, loop=self._loop)<EOL>async def cb(raw_msg):<EOL><INDENT>nonlocal sub<EOL>await sub._msgs_queue.put(raw_msg)<EOL><DEDENT>sid = await self._nc.subscribe(sub.inbox, cb=cb)<EOL>sub.sid = sid<EOL>req = protocol.SubscriptionRequest()<EOL>req.clientID = self._client_id<EOL>req.maxInFlight = max_inflight<EOL>req.ackWaitInSecs = ack_wait<EOL>if queue is not None:<EOL><INDENT>req.qGroup = queue<EOL><DEDENT>if durable_name is not None:<EOL><INDENT>req.durableName = durable_name<EOL><DEDENT>if deliver_all_available:<EOL><INDENT>req.startPosition = protocol.First<EOL><DEDENT>elif start_at is None or start_at == '<STR_LIT>':<EOL><INDENT>req.startPosition = protocol.NewOnly<EOL><DEDENT>elif start_at == '<STR_LIT>':<EOL><INDENT>req.startPosition = protocol.LastReceived<EOL><DEDENT>elif start_at == '<STR_LIT:time>':<EOL><INDENT>req.startPosition = protocol.TimeDeltaStart<EOL>req.startTimeDelta = int(now() - time) * <NUM_LIT><EOL><DEDENT>elif start_at == '<STR_LIT>':<EOL><INDENT>req.startPosition = protocol.SequenceStart<EOL>req.startSequence = sequence<EOL><DEDENT>elif start_at == '<STR_LIT>':<EOL><INDENT>req.startPosition = protocol.First<EOL><DEDENT>req.subject = subject<EOL>req.inbox = sub.inbox<EOL>msg = await self._nc.request(<EOL>self._sub_req_subject,<EOL>req.SerializeToString(),<EOL>self._connect_timeout,<EOL>)<EOL>resp = protocol.SubscriptionResponse()<EOL>resp.ParseFromString(msg.data)<EOL>if resp.error != "<STR_LIT>":<EOL><INDENT>try:<EOL><INDENT>await self._nc.unsubscribe(sid)<EOL><DEDENT>except:<EOL><INDENT>pass<EOL><DEDENT>raise StanError(resp.error)<EOL><DEDENT>sub.ack_inbox = resp.ackInbox<EOL>sub._msgs_task = self._loop.create_task(self._process_msg(sub))<EOL>return sub<EOL>
:param subject: Subject for the NATS Streaming subscription.

:param cb: Callback which will be dispatched the

:param error_cb: Async callback called on error, with the exception as
the sole argument.

:param start_at: One of the following options:
   - 'new_only' (default)
   - 'first'
   - 'sequence'
   - 'last_received'
   - 'time'

:param deliver_all_available: Signals to receive all messages.

:param sequence: Start sequence position from which we will be
receiving the messages.

:param time: Unix timestamp after which the messages will be delivered.

:param manual_acks: Disables auto ack functionality in the subscription
callback so that it is implemented by the user instead.

:param ack_wait: How long to wait for an ack before being redelivered
previous messages.

:param durable_name: Name of the durable subscription.

:param: pending_limits: Max number of messages to await in subscription
before it becomes a slow consumer.