def _send_init_message_and_wait_ack(self) -> None:
        """Send init message to the provided websocket and wait for the connection ACK.

        If the answer is not a connection_ack message, we will return an Exception.
        """

        init_message = json.dumps(
            {"type": "connection_init", "payload": self.init_payload}
        )

        await self._send(init_message)

        # Wait for the connection_ack message or raise a TimeoutError
        init_answer = await asyncio.wait_for(self._receive(), self.ack_timeout)

        answer_type, answer_id, execution_result = self._parse_answer(init_answer)

        if answer_type != "connection_ack":
            raise TransportProtocolError(
                "Websocket server did not return a connection ack"
            ) 

def _clean_close(self, e: Exception) -> None:
        """Coroutine which will:

        - send stop messages for each active subscription to the server
        - send the connection terminate message
        """

        # Send 'stop' message for all current queries
        for query_id, listener in self.listeners.items():

            if listener.send_stop:
                await self._send_stop_message(query_id)
                listener.send_stop = False

        # Wait that there is no more listeners (we received 'complete' for all queries)
        try:
            await asyncio.wait_for(self._no_more_listeners.wait(), self.close_timeout)
        except asyncio.TimeoutError:  # pragma: no cover
            pass

        # Finally send the 'connection_terminate' message
        await self._send_connection_terminate_message() 

def fetch(self, session, room, id_blacklist):
        url = self._stream_api.format(room=room)
        while True:
            # print("polling on url %s" % url)
            try:
                with aiohttp.Timeout(300):
                    async with session.get(url, headers=self.headers) as resp:
                        while True:
                            line = await resp.content.readline()
                            line = bytes.decode(line, 'utf-8').strip()
                            if not line:
                                continue
                            msg = self.parse_jmsg(room, json.loads(line))
                            if msg.sender in id_blacklist:
                                continue
                            self.send_to_bus(msg)
            except asyncio.TimeoutError:
                pass
            except:
                raise 

def _parse_sdcard_list(self, done_cb):
        self.log.debug('Comms: _parse_sdcard_list')

        # setup callback to receive and parse listing data
        files = []
        f = asyncio.Future()
        self.redirect_incoming(lambda x: self._rcv_sdcard_line(x, files, f))

        # issue command
        self._write('M20\n')

        # wait for it to complete and get all the lines
        # add a long timeout in case it fails and we don't want to wait for ever
        try:
            await asyncio.wait_for(f, 10)

        except asyncio.TimeoutError:
            self.log.warning("Comms: Timeout waiting for sd card list")
            files = []

        self.redirect_incoming(None)

        # call upstream callback with results
        done_cb(files) 

def _connect(self, timeout=None, ssl=None):
        await super()._connect(timeout=timeout, ssl=ssl)

        # Wait for EOF for 2 seconds (or if _wait_for_data's definition
        # is missing or different, just sleep for 2 seconds). This way
        # we give the proxy a chance to close the connection if the current
        # codec (which the proxy detects with the data we sent) cannot
        # be used for this proxy. This is a work around for #1134.
        # TODO Sleeping for N seconds may not be the best solution
        # TODO This fix could be welcome for HTTP proxies as well
        try:
            await asyncio.wait_for(self._reader._wait_for_data('proxy'), 2)
        except asyncio.TimeoutError:
            pass
        except Exception:
            await asyncio.sleep(2)

        if self._reader.at_eof():
            await self.disconnect()
            raise ConnectionError(
                'Proxy closed the connection after sending initial payload') 

def _connect_sentinel(self, address, timeout, pools):
        """Try to connect to specified Sentinel returning either
        connections pool or exception.
        """
        try:
            with async_timeout(timeout):
                pool = await create_pool(
                    address, minsize=1, maxsize=2,
                    parser=self._parser_class,
                    )
            pools.append(pool)
            return pool
        except asyncio.TimeoutError as err:
            sentinel_logger.debug(
                "Failed to connect to Sentinel(%r) within %ss timeout",
                address, timeout)
            return err
        except Exception as err:
            sentinel_logger.debug(
                "Error connecting to Sentinel(%r): %r", address, err)
            return err 

def test_create_no_minsize(create_pool, server):
    pool = await create_pool(
        server.tcp_address,
        minsize=0, maxsize=1)
    assert pool.size == 0
    assert pool.freesize == 0

    with (await pool):
        assert pool.size == 1
        assert pool.freesize == 0

        with pytest.raises(asyncio.TimeoutError):
            await asyncio.wait_for(pool.acquire(),
                                   timeout=0.2)
    assert pool.size == 1
    assert pool.freesize == 1 

def simple_db_mutate_returning_item(result_cls, context, mutation_query, *,
                                          item_query, item_cls):
    async with context['dbpool'].acquire() as conn, conn.begin():
        try:
            result = await conn.execute(mutation_query)
            if result.rowcount > 0:
                result = await conn.execute(item_query)
                item = await result.first()
                return result_cls(True, 'success', item_cls.from_row(item))
            else:
                return result_cls(False, 'no matching record', None)
        except (pg.IntegrityError, sa.exc.IntegrityError) as e:
            return result_cls(False, f'integrity error: {e}', None)
        except (asyncio.CancelledError, asyncio.TimeoutError):
            raise
        except Exception as e:
            return result_cls(False, f'unexpected error: {e}', None) 

def on_message(self, message):
        # we do not want the bot to reply to itself
        if message.author.id == self.user.id:
            return

        if message.content.startswith('$guess'):
            await message.channel.send('Guess a number between 1 and 10.')

            def is_correct(m):
                return m.author == message.author and m.content.isdigit()

            answer = random.randint(1, 10)

            try:
                guess = await self.wait_for('message', check=is_correct, timeout=5.0)
            except asyncio.TimeoutError:
                return await message.channel.send('Sorry, you took too long it was {}.'.format(answer))

            if int(guess.content) == answer:
                await message.channel.send('You are right!')
            else:
                await message.channel.send('Oops. It is actually {}.'.format(answer)) 

def request_offline_members(self, guilds):
        # get all the chunks
        chunks = []
        for guild in guilds:
            chunks.extend(self.chunks_needed(guild))

        # we only want to request ~75 guilds per chunk request.
        splits = [guilds[i:i + 75] for i in range(0, len(guilds), 75)]
        for split in splits:
            await self.chunker([g.id for g in split])

        # wait for the chunks
        if chunks:
            try:
                await utils.sane_wait_for(chunks, timeout=len(chunks) * 30.0)
            except asyncio.TimeoutError:
                log.warning('Somehow timed out waiting for chunks.')
            else:
                log.info('Finished requesting guild member chunks for %d guilds.', len(guilds)) 

def request_offline_members(self, guilds, *, shard_id):
        # get all the chunks
        chunks = []
        for guild in guilds:
            chunks.extend(self.chunks_needed(guild))

        # we only want to request ~75 guilds per chunk request.
        splits = [guilds[i:i + 75] for i in range(0, len(guilds), 75)]
        for split in splits:
            await self.chunker([g.id for g in split], shard_id=shard_id)

        # wait for the chunks
        if chunks:
            try:
                await utils.sane_wait_for(chunks, timeout=len(chunks) * 30.0)
            except asyncio.TimeoutError:
                log.info('Somehow timed out waiting for chunks.')
            else:
                log.info('Finished requesting guild member chunks for %d guilds.', len(guilds)) 

def character_delete(self, ctx, *, name):
        """Deletes a character."""
        user_characters = await self.bot.mdb.characters.find(
            {"owner": str(ctx.author.id)}, ['name', 'upstream']
        ).to_list(None)
        if not user_characters:
            return await ctx.send('You have no characters.')

        selected_char = await search_and_select(ctx, user_characters, name, lambda e: e['name'],
                                                selectkey=lambda e: f"{e['name']} (`{e['upstream']}`)")

        await ctx.send(f"Are you sure you want to delete {selected_char['name']}? (Reply with yes/no)")
        try:
            reply = await self.bot.wait_for('message', timeout=30, check=auth_and_chan(ctx))
        except asyncio.TimeoutError:
            reply = None
        reply = get_positivity(reply.content) if reply is not None else None
        if reply is None:
            return await ctx.send('Timed out waiting for a response or invalid response.')
        elif reply:
            await Character.delete(ctx, str(ctx.author.id), selected_char['upstream'])
            return await ctx.send(f"{selected_char['name']} has been deleted.")
        else:
            return await ctx.send("OK, cancelling.") 

def transferchar(self, ctx, user: discord.Member):
        """Gives a copy of the active character to another user."""
        character: Character = await Character.from_ctx(ctx)
        overwrite = ''

        conflict = await self.bot.mdb.characters.find_one({"owner": str(user.id), "upstream": character.upstream})
        if conflict:
            overwrite = "**WARNING**: This will overwrite an existing character."

        await ctx.send(f"{user.mention}, accept a copy of {character.name}? (Type yes/no)\n{overwrite}")
        try:
            m = await self.bot.wait_for('message', timeout=300,
                                        check=lambda msg: msg.author == user
                                                          and msg.channel == ctx.channel
                                                          and get_positivity(msg.content) is not None)
        except asyncio.TimeoutError:
            m = None

        if m is None or not get_positivity(m.content): return await ctx.send("Transfer not confirmed, aborting.")

        character.owner = str(user.id)
        await character.commit(ctx)
        await ctx.send(f"Copied {character.name} to {user.display_name}'s storage.") 

def confirm(ctx, message, delete_msgs=False):
    """
    Confirms whether a user wants to take an action.
    :rtype: bool|None
    :param ctx: The current Context.
    :param message: The message for the user to confirm.
    :param delete_msgs: Whether to delete the messages.
    :return: Whether the user confirmed or not. None if no reply was recieved
    """
    msg = await ctx.channel.send(message)
    try:
        reply = await ctx.bot.wait_for('message', timeout=30, check=auth_and_chan(ctx))
    except asyncio.TimeoutError:
        return None
    replyBool = get_positivity(reply.content) if reply is not None else None
    if delete_msgs:
        try:
            await msg.delete()
            await reply.delete()
        except:
            pass
    return replyBool 

def read_messages(self, stream_name: str, options: Optional[ReadMessagesOptions] = None) -> List[Message]:
        """
        Read message(s) from a chosen stream with options. If no options are specified it will try to read
        1 message from the stream.

        :param stream_name: The name of the stream to read from.
        :param options: (Optional) Options used when reading from the stream of type :class:`.data.ReadMessagesOptions`.
            Defaults are:

            * desired_start_sequence_number: 0,
            * min_message_count: 1,
            * max_message_count: 1,
            * read_timeout_millis: 0 ``# Where 0 here represents that the server will immediately return the messages``
                                     ``# or an exception if there were not enough messages available.``
            If desired_start_sequence_number is specified in the options and is less
            than the current beginning of the stream, returned messages will start
            at the beginning of the stream and not necessarily the desired_start_sequence_number.
        :return: List of at least 1 message.
        :raises: :exc:`~.exceptions.StreamManagerException` and subtypes based on the precise error.
        :raises: :exc:`asyncio.TimeoutError` if the request times out.
        :raises: :exc:`ConnectionError` if the client is unable to reconnect to the server.
        """
        self.__check_closed()
        return Util.sync(self._read_messages(stream_name, options), loop=self.__loop) 

def test_unclosed_client_session_issue_645_in_async_mode(self):
        def exception_handler(_, context):
            nonlocal session_unclosed
            if context["message"] == "Unclosed client session":
                session_unclosed = True

        async def issue_645():
            client = WebClient(base_url="http://localhost:8888", timeout=1, run_async=True)
            try:
                await client.users_list(token="xoxb-timeout")
            except asyncio.TimeoutError:
                pass

        session_unclosed = False
        loop = asyncio.get_event_loop()
        loop.set_exception_handler(exception_handler)
        loop.run_until_complete(issue_645())
        gc.collect()  # force Python to gc unclosed client session
        self.assertFalse(session_unclosed, "Unclosed client session") 

def work_async(self):
        """Perform a single Connection iteration asynchronously."""
        try:
            raise self._error
        except TypeError:
            pass
        except Exception as e:
            _logger.warning("%r", e)
            raise
        try:
            await self.lock_async()
            if self._closing:
                _logger.debug("Connection unlocked but shutting down.")
                return
            await asyncio.sleep(0, loop=self.loop)
            self._conn.do_work()
        except asyncio.TimeoutError:
            _logger.debug("Connection %r timed out while waiting for lock acquisition.", self.container_id)
        finally:
            await asyncio.sleep(0, loop=self.loop)
            self.release_async() 

def redirect_async(self, redirect_error, auth):
        """Redirect the connection to an alternative endpoint.
        :param redirect: The Link DETACH redirect details.
        :type redirect: ~uamqp.errors.LinkRedirect
        :param auth: Authentication credentials to the redirected endpoint.
        :type auth: ~uamqp.authentication.common.AMQPAuth
        """
        _logger.info("Redirecting connection %r.", self.container_id)
        try:
            await self.lock_async()
            if self.hostname == redirect_error.hostname:
                return
            if self._state != c_uamqp.ConnectionState.END:
                await self._close_async()
            self.hostname = redirect_error.hostname
            self.auth = auth
            self._conn = self._create_connection(auth)
            for setting, value in self._settings.items():
                setattr(self, setting, value)
            self._error = None
            self._closing = False
        except asyncio.TimeoutError:
            _logger.debug("Connection %r timed out while waiting for lock acquisition.", self.container_id)
        finally:
            self.release_async() 

def destroy_async(self):
        """Close the connection asynchronously, and close any associated
        CBS authentication session.
        """
        try:
            await self.lock_async()
            _logger.debug("Unlocked connection %r to close.", self.container_id)
            await self._close_async()
        except asyncio.TimeoutError:
            _logger.debug(
                "Connection %r timed out while waiting for lock acquisition on destroy. Destroying anyway.",
                self.container_id)
            await self._close_async()
        finally:
            self.release_async()
        uamqp._Platform.deinitialize()  # pylint: disable=protected-access 

def wait(self, timeout=None):
        deadline = Deadline(timeout)
        barrier_lifted = self.client.wait_for_events(
            [WatchEvent.DELETED], self.path
        )

        exists = await self.client.exists(path=self.path, watch=True)
        if not exists:
            return

        try:
            if not deadline.is_indefinite:
                await asyncio.wait_for(barrier_lifted, deadline.timeout)
            else:
                await barrier_lifted
        except asyncio.TimeoutError:
            raise exc.TimeoutError 

def test_websocket_server_does_not_send_ack(event_loop, server, query_str):

    url = f"ws://{server.hostname}:{server.port}/graphql"

    sample_transport = WebsocketsTransport(url=url, ack_timeout=1)

    with pytest.raises(asyncio.TimeoutError):
        async with Client(transport=sample_transport):
            pass 

def stop(self):
        print("Stopping server")

        self.server.close()
        try:
            await asyncio.wait_for(self.server.wait_closed(), timeout=1)
        except asyncio.TimeoutError:  # pragma: no cover
            assert False, "Server failed to stop"

        print("Server stopped\n\n\n") 

def get_data(self, raw: bool = True) -> AnyStr:
        """The request body data."""
        try:
            body_future = asyncio.ensure_future(self.body)
            raw_data = await asyncio.wait_for(body_future, timeout=self.body_timeout)
        except asyncio.TimeoutError:
            body_future.cancel()
            from ..exceptions import RequestTimeout  # noqa Avoiding circular import

            raise RequestTimeout()

        if raw:
            return raw_data
        else:
            return raw_data.decode(self.charset) 

def handle_request(self, request: Request, send: Callable) -> None:
        try:
            response = await self.app.handle_request(request)
        except Exception:
            response = await traceback_response()

        if response.timeout != sentinel:
            timeout = cast(Optional[float], response.timeout)
        else:
            timeout = self.app.config["RESPONSE_TIMEOUT"]
        try:
            await asyncio.wait_for(self._send_response(send, response), timeout=timeout)
        except asyncio.TimeoutError:
            pass 

def _arequest(self, url, **params):
        method = params.get('method', 'GET')
        json_data = params.get('json', {})
        timeout = params.pop('timeout', None) or self.timeout
        try:
            async with self.session.request(
                method, url, timeout=timeout, headers=self.headers, params=params, data=json_data
            ) as resp:
                return self._raise_for_status(resp, await resp.text())
        except asyncio.TimeoutError:
            raise NotResponding
        except aiohttp.ServerDisconnectedError:
            raise NetworkError 

def _arequest(self, url, **params):
        timeout = params.pop('timeout', None) or self.timeout
        try:
            async with self.session.get(url, timeout=timeout, headers=self.headers, params=params) as resp:
                return self._raise_for_status(resp, await resp.text())
        except asyncio.TimeoutError:
            raise NotResponding
        except aiohttp.ServerDisconnectedError:
            raise NetworkError 

def get_info(app, url):
    """
    Invoke the /info request on the indicated url and return the response
    """
    req = url + "/info"
    log.info(f"get_info({url})")
    try:
        log.debug("about to call http_get")
        rsp_json = await http_get(app, req)
        log.debug("called http_get")
        if "node" not in rsp_json:
            log.error("Unexpected response from node")
            return None

    except OSError as ose:
        log.warn(f"OSError for req: {req}: {ose}")
        return None

    except HTTPInternalServerError as hpe:
        log.warn(f"HTTPInternalServerError for req {req}: {hpe}")
        # node has gone away?
        return None

    except HTTPNotFound as nfe:
        log.warn(f"HTTPNotFound error for req {req}: {nfe}")
        # node has gone away?
        return None

    except TimeoutError as toe:
        log.warn(f"Timeout error for req: {req}: {toe}")
        # node has gone away?
        return None
    except HTTPGone as hg:
        log.warn("Timeout error for req: {}: {}".format(req, str(hg)))
        # node has gone away?
        return None
    except:
        log.warn("uncaught exception in get_info")

    return rsp_json 

def _try_connect(self, attempt):
        try:
            self._log.debug('Connection attempt %d...', attempt)
            await self._connection.connect(timeout=self._connect_timeout)
            self._log.debug('Connection success!')
            return True
        except (IOError, asyncio.TimeoutError) as e:
            self._log.warning('Attempt %d at connecting failed: %s: %s',
                              attempt, type(e).__name__, e)
            await asyncio.sleep(self._delay)
            return False 

def test_wait_for_confirmation(self):
        """The message should always be edited and only return True if the emoji is a check mark."""
        subtests = (
            (constants.Emojis.check_mark, True, None),
            ("InVaLiD", False, None),
            (None, False, asyncio.TimeoutError),
        )

        for emoji, ret_val, side_effect in subtests:
            for bot in (True, False):
                with self.subTest(emoji=emoji, ret_val=ret_val, side_effect=side_effect, bot=bot):
                    # Set up mocks
                    message = helpers.MockMessage()
                    member = helpers.MockMember(bot=bot)

                    self.bot.wait_for.reset_mock()
                    self.bot.wait_for.return_value = (helpers.MockReaction(emoji=emoji), None)
                    self.bot.wait_for.side_effect = side_effect

                    # Call the function
                    actual_return = await self.syncer._wait_for_confirmation(member, message)

                    # Perform assertions
                    self.bot.wait_for.assert_called_once()
                    self.assertIn("reaction_add", self.bot.wait_for.call_args[0])

                    message.edit.assert_called_once()
                    kwargs = message.edit.call_args[1]
                    self.assertIn("content", kwargs)

                    # Core devs should only be mentioned if the author is a bot.
                    if bot:
                        self.assertIn(self.syncer._CORE_DEV_MENTION, kwargs["content"])
                    else:
                        self.assertNotIn(self.syncer._CORE_DEV_MENTION, kwargs["content"])

                    self.assertIs(actual_return, ret_val) 

def test_continue_eval_does_not_continue(self):
        ctx = MockContext(message=MockMessage(clear_reactions=AsyncMock()))
        self.bot.wait_for.side_effect = asyncio.TimeoutError

        actual = await self.cog.continue_eval(ctx, MockMessage())
        self.assertEqual(actual, None)
        ctx.message.clear_reactions.assert_called_once()