KaneAI - World’s First E2E Software Testing Agent.Book a Demo Join TestMu Conference

How to use _do_close method in localstack

Best Python code snippet using localstack_python

connection.py

Source:connection.py

...35            self._closing = True36            self._loop.call_soon(self._do_close, None)37        return self._closed38    def close(self):39        return self._do_close()40    def _do_close(self, exc=None):41        if self._closed:42            return43        logger.info('closing connection..')44        self._writer.close()45        self._read_task.cancel()46        self._reader = None47        self._writer = None48        self._closing = False49        self._closed = True50        while self._queue:51            command, waiter = self._queue.popleft()52            logger.debug('cancelling waiter %r', (command, waiter))53            if exc is None:54                waiter.cancel()55            else:56                waiter.set_exception(exc)57    async def wait_closed(self):58        return await asyncio.shield(self._close_waiter, loop=self._loop)59    def execute(self, command, *args, body=None):60        if self.closed:61            raise ConnectionClosedError(62                'cannot execute command because the connection is closed')63        waiter = self._loop.create_future()64        self._queue.append((command, waiter))65        for part in encode_command(command, *args, body=body):66            self._writer.write(part)67            logger.debug('scheduled to write %s', part[:30])68        return waiter69    async def _read_loop(self):70        exc = None71        try:72            while not self._reader.at_eof():73                head = await self._reader.readline()74                logger.debug('read: head %s', head)75                if head == b'' or self._reader.at_eof():76                    break77                status, headers, body_len = handle_head(head)78                if body_len:79                    body = await self._reader.readexactly(body_len)80                    await self._reader.readexactly(2)  # crlf81                    logger.debug('read: body %s', body[:30])82                else:83                    body = None84                # handler response85                try:86                    command, waiter = self._queue.popleft()87                    result = handle_response(command, status, headers, body)88                except Exception as e:89                    waiter.set_exception(e)90                else:91                    waiter.set_result(result)92        except Exception as e:93            if not isinstance(e, (CancelledError, IncompleteReadError)):94                exc = e95        self._closing = True...

connections.py

Source:connections.py

...30        return len(self)31    def close(self, conn):32        if self.contains(conn):33            self._conns.remove(conn)34        self._do_close(conn)35    def closeAll(self):36        for conn in self._conns:37            self._do_close(conn)38        self._conns = []39    def _do_close(self, conn):40        if hasattr(conn, 'close'):41            conn.close()42        else:43            conn.REQUEST.close()44            conn._p_jar.close()45registry = ConnectionRegistry()46register = registry.register47contains = registry.contains48count = registry.count49close = registry.close...

Automation Testing Tutorials

Learn to execute automation testing from scratch with LambdaTest Learning Hub. Right from setting up the prerequisites to run your first automation test, to following best practices and diving deeper into advanced test scenarios. LambdaTest Learning Hubs compile a list of step-by-step guides to help you be proficient with different test automation frameworks i.e. Selenium, Cypress, TestNG etc.