async def connect(self, cluster_id, client_id,<EOL>nats=None,<EOL>connect_timeout=DEFAULT_CONNECT_TIMEOUT,<EOL>max_pub_acks_inflight=DEFAULT_MAX_PUB_ACKS_INFLIGHT,<EOL>loop=None,<EOL>):
self._cluster_id = cluster_id<EOL>self._client_id = client_id<EOL>self._loop = loop<EOL>self._connect_timeout = connect_timeout<EOL>if nats is not None:<EOL><INDENT>self._nc = nats<EOL>self._loop = self._nc._loop<EOL><DEDENT>self._discover_subject = DEFAULT_DISCOVER_SUBJECT % self._cluster_id<EOL>self._hb_inbox = new_guid()<EOL>self._ack_subject = DEFAULT_ACKS_SUBJECT % new_guid()<EOL>self._pending_pub_acks_queue = asyncio.Queue(<EOL>maxsize=max_pub_acks_inflight, loop=self._loop)<EOL>self._hb_inbox_sid = await self._nc.subscribe(<EOL>self._hb_inbox,<EOL>cb=self._process_heartbeats,<EOL>)<EOL>self._ack_subject_sid = await self._nc.subscribe(<EOL>self._ack_subject,<EOL>cb=self._process_ack,<EOL>)<EOL>await self._nc.flush()<EOL>creq = protocol.ConnectRequest()<EOL>creq.clientID = self._client_id<EOL>creq.heartbeatInbox = self._hb_inbox<EOL>payload = creq.SerializeToString()<EOL>msg = None<EOL>try:<EOL><INDENT>msg = await self._nc.request(<EOL>self._discover_subject,<EOL>payload,<EOL>timeout=self._connect_timeout,<EOL>)<EOL><DEDENT>except:<EOL><INDENT>await self._close()<EOL>raise ErrConnectReqTimeout("<STR_LIT>".format(cluster_id))<EOL><DEDENT>resp = protocol.ConnectResponse()<EOL>resp.ParseFromString(msg.data)<EOL>if resp.error != "<STR_LIT>":<EOL><INDENT>try:<EOL><INDENT>await self._close()<EOL><DEDENT>except:<EOL><INDENT>pass<EOL><DEDENT>raise StanError(resp.error)<EOL><DEDENT>self._pub_prefix = resp.pubPrefix<EOL>self._sub_req_subject = resp.subRequests<EOL>self._unsub_req_subject = resp.unsubRequests<EOL>self._close_req_subject = resp.closeRequests<EOL>self._sub_close_req_subject = resp.subCloseRequests<EOL>
Starts a session with a NATS Streaming cluster.

:param cluster: Name of the cluster to which we will connect.
:param nats: NATS connection to be borrowed for NATS Streaming.