nats/aio/client.py
def drain(self, sid=None):
        """
        Drain will put a connection into a drain state. All subscriptions will
        immediately be put into a drain state. Upon completion, the publishers
        will be drained and can not publish any additional messages. Upon draining
        of the publishers, the connection will be closed. Use the `closed_cb'
        option to know when the connection has moved from draining to closed.

        If a sid is passed, just the subscription with that sid will be drained
        without closing the connection.
        """
        if self.is_draining:
            return
        if self.is_closed:
            raise ErrConnectionClosed
        if self.is_connecting or self.is_reconnecting:
            raise ErrConnectionReconnecting

        if sid is not None:
            return self._drain_sub(sid)

        # Start draining the subscriptions
        self._status = Client.DRAINING_SUBS

        drain_tasks = []
        for ssid, sub in self._subs.items():
            task = self._drain_sub(ssid)
            drain_tasks.append(task)

        drain_is_done = asyncio.gather(*drain_tasks)
        try:
            yield from asyncio.wait_for(drain_is_done, self.options["drain_timeout"])
        except asyncio.TimeoutError:
            drain_is_done.exception()
            drain_is_done.cancel()
            if self._error_cb is not None:
                yield from self._error_cb(ErrDrainTimeout)
        except asyncio.CancelledError:
            pass
        finally:
            self._status = Client.DRAINING_PUBS
            yield from self.flush()
            yield from self._close(Client.CLOSED)
def drain(self, sid=None):
        """
        Drain will put a connection into a drain state. All subscriptions will
        immediately be put into a drain state. Upon completion, the publishers
        will be drained and can not publish any additional messages. Upon draining
        of the publishers, the connection will be closed. Use the `closed_cb'
        option to know when the connection has moved from draining to closed.

        If a sid is passed, just the subscription with that sid will be drained
        without closing the connection.
        """
        if self.is_draining:
            return
        if self.is_closed:
            raise ErrConnectionClosed
        if self.is_connecting or self.is_reconnecting:
            raise ErrConnectionReconnecting

        if sid is not None:
            return self._drain_sub(sid)

        # Start draining the subscriptions
        self._status = Client.DRAINING_SUBS

        drain_tasks = []
        for ssid, sub in self._subs.items():
            task = self._drain_sub(ssid)
            drain_tasks.append(task)

        drain_is_done = asyncio.gather(*drain_tasks)
        try:
            yield from asyncio.wait_for(drain_is_done, self.options["drain_timeout"])
        except asyncio.TimeoutError:
            drain_is_done.exception()
            drain_is_done.cancel()
            if self._error_cb is not None:
                yield from self._error_cb(ErrDrainTimeout)
        except asyncio.CancelledError:
            pass
        finally:
            self._status = Client.DRAINING_PUBS
            yield from self.flush()
            yield from self._close(Client.CLOSED)
Drain will put a connection into a drain state. All subscriptions will
        immediately be put into a drain state. Upon completion, the publishers
        will be drained and can not publish any additional messages. Upon draining
        of the publishers, the connection will be closed. Use the `closed_cb'
        option to know when the connection has moved from draining to closed.

        If a sid is passed, just the subscription with that sid will be drained
        without closing the connection.