def _wait_until_up(self):
        async with aiohttp.ClientSession() as session:
            for i in range(0, 30):
                if self.exitcode is not None:
                    pytest.fail('unable to start/connect to aiohttp server')
                    return

                try:
                    # we need to bypass the proxies due to monkey patches
                    await session.get(self.endpoint_url + '/ok', timeout=0.5)
                    return
                except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
                    await asyncio.sleep(0.5)
                except BaseException:
                    pytest.fail('unable to start/connect to aiohttp server')
                    raise

        pytest.fail('unable to start and connect to aiohttp server') 

def _start(self):
        self._thread = threading.Thread(target=self._server_entry, daemon=True)
        self._thread.start()

        async with aiohttp.ClientSession() as session:
            start = time.time()

            while time.time() - start < 10:
                if not self._thread.is_alive():
                    break

                try:
                    # we need to bypass the proxies due to monkeypatches
                    async with session.get(self.endpoint_url + '/static',
                                           timeout=_CONNECT_TIMEOUT):
                        pass
                    break
                except (asyncio.TimeoutError, aiohttp.ClientConnectionError):
                    await asyncio.sleep(0.5)
            else:
                await self._stop()  # pytest.fail doesn't call stop_process
                raise Exception("Can not start service: {}".format(
                    self._service_name)) 

def test_client_connection_failure(rpc_context, unused_tcp_port_factory):
    client = JsonRpcClient(
        url='ws://{host}:{port}{url}'.format(
            host=rpc_context.host, port=rpc_context.port,
            url=rpc_context.url,
        )
    )

    with pytest.raises(aiohttp.ClientConnectionError):
        await client.connect_url(
            'ws://{host}:{port}{url}'.format(
                host=rpc_context.host, port=unused_tcp_port_factory(),
                url=rpc_context.url,
            )
        )
    assert client._session.closed is True 

def fetch(self, url, **kwargs) -> aiohttp.ClientResponse:
        headers = {"User-Agent": get_user_agent()}
        async with aiohttp.ClientSession(
            conn_timeout=self.config["response_timeout"],
            read_timeout=self.config["response_timeout"],
        ) as session:
            try:
                async with session.get(
                    urljoin(self.host, url), headers=headers, **kwargs
                ) as response:
                    await response.text()
                    return response
            except aiohttp.ClientConnectionError:
                raise exceptions.SlaveDoesNotExist(
                    f"Unable to connect to the slave at {self.host}"
                ) 

def _sync(self) -> None:
        if not self.is_real_user:
            self.log.warning("Called sync() for non-custom puppet.")
            return
        custom_mxid: UserID = self.custom_mxid
        access_token_at_start: str = self.access_token
        errors: int = 0
        filter_id: FilterID = await self._create_sync_filter()
        self.log.debug(f"Starting syncer for {custom_mxid} with sync filter {filter_id}.")
        while access_token_at_start == self.access_token:
            try:
                cur_batch = self.next_batch
                sync_resp = await self.intent.sync(filter_id=filter_id, since=cur_batch,
                                                   set_presence=PresenceState.OFFLINE)
                self.next_batch = sync_resp.get("next_batch", None)
                errors = 0
                if cur_batch is not None:
                    self._handle_sync(sync_resp)
            except (MatrixError, ClientConnectionError, asyncio.TimeoutError) as e:
                errors += 1
                wait = min(errors, 11) ** 2
                self.log.warning(f"Syncer for {custom_mxid} errored: {e}. "
                                 f"Waiting for {wait} seconds...")
                await asyncio.sleep(wait)
        self.log.debug(f"Syncer for custom puppet {custom_mxid} stopped.") 

def fetch(self):
        while True:

            try:
                hdrlen = constants.STREAM_HEADER_SIZE_BYTES
                header = yield from self._response.content.readexactly(hdrlen)

                _, length = struct.unpack(">BxxxL", header)
                if not length:
                    continue

                data = yield from self._response.content.readexactly(length)

            except (
                aiohttp.ClientConnectionError,
                aiohttp.ServerDisconnectedError,
                asyncio.IncompleteReadError,
            ):
                break
            return data 

def post(self, url: str, data: Dict[str, str] = None) -> Response:
        """Perform HTTP POST request.

        :param url: the request url
        :param data: the data send to server
        :return: the response from server
        :raise: :exc:`ConnectionError <stellar_sdk.exceptions.ConnectionError>`
        """
        try:
            response = await self._session.post(url, data=data, timeout=aiohttp.ClientTimeout(total=self.post_timeout))
            return Response(
                status_code=response.status,
                text=await response.text(),
                headers=dict(response.headers),
                url=str(response.url),
            )
        except aiohttp.ClientConnectionError as e:
            raise ConnectionError(e) 

def make_request(self, method, params=None, data=None, **kwargs):
        headers = {
            "accept": "application/json",
            "Connection": "keep-alive",
            "Content-Type": "application/json"
        }
        kwargs.setdefault('headers', headers)
        kwargs.setdefault('timeout', 10)
        error = None
        try:
            response = await method(self._endpoint, params=params, data=data, **kwargs)
            return await response.json()
        except aiohttp.ClientConnectionError as e:
            print("Unable to connect to Thor-Restful server:")
            error = e
        except Exception as e:
            try:
                text = await response.text()
                error = Exception(text.strip('\n'))
            except:
                error = e
        print("Thor-Restful server Err:")
        raise error 

def test_exceptions(self):
        import aiohttp

        exceptions = aiohttp_.AiohttpClient.exceptions

        with pytest.raises(exceptions.BaseClientException):
            raise aiohttp.ClientError()

        with pytest.raises(exceptions.BaseClientException):
            # Test polymorphism
            raise aiohttp.InvalidURL("invalid")

        with pytest.raises(exceptions.ConnectionError):
            raise aiohttp.ClientConnectionError()

        with pytest.raises(exceptions.ConnectionTimeout):
            raise aiohttp.ClientConnectorError.__new__(
                aiohttp.ClientConnectorError
            )

        with pytest.raises(exceptions.ServerTimeout):
            raise aiohttp.ServerTimeoutError()

        with pytest.raises(exceptions.SSLError):
            raise aiohttp.ClientSSLError.__new__(aiohttp.ClientSSLError)

        with pytest.raises(exceptions.InvalidURL):
            raise aiohttp.InvalidURL("invalid") 

def test_stream_reconnection_client_connection_error():
    async with Stream() as stream:
        async def client_connection_error():
            raise aiohttp.ClientConnectionError

        with patch.object(stream, '_connect', side_effect=stream_content):
            data = await stream.__anext__()
            assert 'connected' in data
            with patch.object(stream.response, 'readline',
                              side_effect=client_connection_error):
                data = await stream.__anext__()
                assert data == {'reconnecting_in': ERROR_TIMEOUT,
                                'error': None} 

def _check_url_async(url: str, session: ClientSession) -> UrlResult:
    """
    Connect to URL and return response status.

    Parameters
    ----------
    url : str
        URL to check
    session : ClientSession
        aiohttp client session

    Returns
    -------
    UrlResult
        Tuple of status code, redirect history, requested url,
        status/error message.

    """
    try:
        async with session.get(url) as resp:
            try:
                await resp.read()
                if resp.history:
                    result = UrlResult(
                        resp.status,
                        resp.history,
                        url,
                        "No error. Redirect to " + str(resp.url),
                    )
                elif resp.status == 200:
                    result = UrlResult(
                        resp.status, resp.history, url, "No error. No redirect."
                    )
                else:
                    result = UrlResult(resp.status, resp.history, url, "Error?")
            except ClientResponseError as client_err:
                return UrlResult(client_err.status, [], url, client_err)
    except ClientConnectionError as err:
        result = UrlResult(404, [], url, err)
    return result 

def _group_send(self, dst_url, messages):
        with self._sentry_group_message_scope(dst_url):
            LOG.info(f'send {len(messages)} messages to {dst_url}')
            data = ActorMessage.batch_encode(messages, self.content_encoding)
            try:
                async with self.session.post(dst_url, data=data, headers=self.headers) as r:
                    await r.read()
            except aiohttp.ClientConnectionError as ex:
                LOG.warning(f'failed to send message to {dst_url}: {ex}')
                return
            except aiohttp.ClientError as ex:
                LOG.warning(f'failed to send message to {dst_url}: {ex}')
                raise
            aiohttp_raise_for_status(r) 

def handle_exception():
    """
    Context manager translating network related exceptions
    to custom :mod:`~galaxy.api.errors`.
    """
    try:
        yield
    except asyncio.TimeoutError:
        raise BackendTimeout()
    except aiohttp.ServerDisconnectedError:
        raise BackendNotAvailable()
    except aiohttp.ClientConnectionError:
        raise NetworkError()
    except aiohttp.ContentTypeError:
        raise UnknownBackendResponse()
    except aiohttp.ClientResponseError as error:
        if error.status == HTTPStatus.UNAUTHORIZED:
            raise AuthenticationRequired()
        if error.status == HTTPStatus.FORBIDDEN:
            raise AccessDenied()
        if error.status == HTTPStatus.SERVICE_UNAVAILABLE:
            raise BackendNotAvailable()
        if error.status == HTTPStatus.TOO_MANY_REQUESTS:
            raise TooManyRequests()
        if error.status >= 500:
            raise BackendError()
        if error.status >= 400:
            logging.warning(
                "Got status %d while performing %s request for %s",
                error.status, error.request_info.method, str(error.request_info.url)
            )
            raise UnknownError()
    except aiohttp.ClientError:
        logging.exception("Caught exception while performing request")
        raise UnknownError() 

def execute_command(conf, method, params, callback=display):
    async with aiohttp.ClientSession() as session:
        try:
            message = {'method': method, 'params': params}
            async with session.get(conf.api_connection_url, json=message) as resp:
                try:
                    data = await resp.json()
                    if 'result' in data:
                        return callback(data['result'])
                    elif 'error' in data:
                        return callback(data['error'])
                except Exception as e:
                    log.exception('Could not process response from server:', exc_info=e)
        except aiohttp.ClientConnectionError:
            print("Could not connect to daemon. Are you sure it's running?") 

def __anext__(self):
        while True:
            try:
                data = yield from self._response.content.readline()
                if not data:
                    break
            except (aiohttp.ClientConnectionError, aiohttp.ServerDisconnectedError):
                break
            return self._transform(json.loads(data.decode("utf8")))

        raise StopAsyncIteration 

def run(self, **params):
        if self.response:
            warnings.warn("already running", RuntimeWarning, stackelevel=2)
            return
        forced_params = {"follow": True}
        default_params = {"stdout": True, "stderr": True}
        params = ChainMap(forced_params, params, default_params)
        try:
            self.response = await self.docker._query(
                "containers/{self.container._id}/logs".format(self=self), params=params
            )
            while True:
                msg = await self.response.content.readline()
                if not msg:
                    break
                await self.channel.publish(msg)
        except (aiohttp.ClientConnectionError, aiohttp.ServerDisconnectedError):
            pass
        finally:
            # signal termination to subscribers
            await self.channel.publish(None)
            try:
                await self.response.release()
            except Exception:
                pass
            self.response = None 

def test_get_job_by_id_service_unavailable(self):
        """
        Por enquanto deixamos o erro ser propagado.
        """
        get_job_by_id_mock = CoroutineMock(
            side_effect=aiohttp.ClientConnectionError()
        )
        self.backend.client = CoroutineMock(spec=ChronosClient)
        self.backend.client.get_job_by_id = get_job_by_id_mock

        user = User(**USER_WITH_MULTIPLE_ACCOUNTS_DICT)
        account = Account(**ACCOUNT_DEV_DICT)

        with self.assertRaises(aiohttp.ClientConnectionError):
            await self.backend.get_job_by_id("job-id", user, account) 

def init_room(self):
        try:
            async with self._session.get(ROOM_INIT_URL, params={'id': self._tmp_room_id},
                                         ssl=self._ssl) as res:
                if res.status != 200:
                    logger.warning('room %d room_init失败：%d %s', self._tmp_room_id,
                                   res.status, res.reason)
                    return False
                data = await res.json()
                if data['code'] != 0:
                    logger.warning('room %d room_init失败：%s', self._tmp_room_id, data['msg'])
                    return False
                if not self._parse_room_init(data['data']):
                    return False
        except aiohttp.ClientConnectionError:
            logger.exception('room %d room_init失败：', self._tmp_room_id)
            return False

        try:
            async with self._session.get(DANMAKU_SERVER_CONF_URL, params={'room_id': self._tmp_room_id},
                                         ssl=self._ssl) as res:
                if res.status != 200:
                    logger.warning('room %d getConf失败：%d %s', self._tmp_room_id,
                                   res.status, res.reason)
                    return False
                data = await res.json()
                if data['code'] != 0:
                    logger.warning('room %d getConf失败：%s', self._tmp_room_id, data['msg'])
                    return False
                self._host_server_list = data['data']['host_server_list']
                self._host_server_token = data['data']['token']
                if not self._host_server_list:
                    logger.warning('room %d getConf失败：host_server_list为空')
                    return False
        except aiohttp.ClientConnectionError:
            logger.exception('room %d getConf失败：', self._tmp_room_id)
            return False
        return True 

def _heartbeat_loop(self):
        while True:
            try:
                await self._websocket.send_bytes(self._make_packet({}, Operation.HEARTBEAT))
                await asyncio.sleep(self._heartbeat_interval)

            except (asyncio.CancelledError, aiohttp.ClientConnectionError):
                break 

def _http_ready_func(self, p):
        url = 'http://localhost:{}'.format(self.port)
        async with aiohttp.ClientSession() as session:
            try:
                async with session.get(url) as resp:
                    # We only care if we get back *any* response, not just 200
                    # If there's an error response, that can be shown directly to the user
                    self.log.debug('Got code {} back from {}'.format(resp.status, url))
                    return True
            except aiohttp.ClientConnectionError:
                self.log.debug('Connection to {} refused'.format(url))
                return False 

def run(self, wait: int = DEFAULT_WAIT):
        """ Run user polling forever
        Can be manually stopped with:
        >> user.stop()
        """
        self.wait = wait
        logger.info("Polling will be started. Is it OK?")

        await self.get_server()
        await self.on.dispatch()
        self.middleware.add_middleware(self.on.pre_p)
        self.status.dispatched = True

        logger.debug("User Polling successfully started")

        while not self._stop:
            try:
                event = await self.make_long_request(self.long_poll_server)
                if isinstance(event, dict) and event.get("ts"):
                    self.loop.create_task(self.emulate(event))
                    self.long_poll_server["ts"] = event["ts"]
                else:
                    await self.get_server()

            except (
                aiohttp.ClientConnectionError,
                aiohttp.ServerTimeoutError,
                TimeoutError,
            ):
                # No internet connection
                logger.warning("Server Timeout Error!")

            except:
                logger.error(
                    "While user lp was running error occurred \n\n{}".format(
                        traceback.format_exc()
                    )
                )

        logger.error("Polling was stopped") 

def handle_exception():
    """
    Context manager translating network related exceptions
    to custom :mod:`~galaxy.api.errors`.
    """
    try:
        yield
    except asyncio.TimeoutError:
        raise BackendTimeout()
    except aiohttp.ServerDisconnectedError:
        raise BackendNotAvailable()
    except aiohttp.ClientConnectionError:
        raise NetworkError()
    except aiohttp.ContentTypeError as error:
        raise UnknownBackendResponse(error.message)
    except aiohttp.ClientResponseError as error:
        if error.status == HTTPStatus.UNAUTHORIZED:
            raise AuthenticationRequired(error.message)
        if error.status == HTTPStatus.FORBIDDEN:
            raise AccessDenied(error.message)
        if error.status == HTTPStatus.SERVICE_UNAVAILABLE:
            raise BackendNotAvailable(error.message)
        if error.status == HTTPStatus.TOO_MANY_REQUESTS:
            raise TooManyRequests(error.message)
        if error.status >= 500:
            raise BackendError(error.message)
        if error.status >= 400:
            logger.warning(
                "Got status %d while performing %s request for %s",
                error.status, error.request_info.method, str(error.request_info.url)
            )
            raise UnknownError(error.message)
    except aiohttp.ClientError as e:
        logger.exception("Caught exception while performing request")
        raise UnknownError(repr(e)) 

def request(req, **user_kw):
    fn, args, kwargs, timeout, cleanup = _transform(req, **user_kw)

    kwargs.update(_proxy_kwargs())
    try:
        if timeout is None:
            async with fn(*args, **kwargs) as r:
                return await _parse(r)
        else:
            try:
                with async_timeout.timeout(timeout):
                    async with fn(*args, **kwargs) as r:
                        return await _parse(r)

            except asyncio.TimeoutError:
                raise exception.TelegramError('Response timeout', 504, {})

    except aiohttp.ClientConnectionError:
        raise exception.TelegramError('Connection Error', 400, {})

    finally:
        if cleanup:  # e.g. closing one-time session
            if asyncio.iscoroutinefunction(cleanup):
                await cleanup()
            else:
                cleanup() 

def api_call(self, path, body=None, full_url=False):
        """Make the actual call to the HMIP server.

        Throws `HmipWrongHttpStatusError` or `HmipConnectionError` if connection has failed or
        response is not correct."""
        result = None
        if not full_url:
            path = self.full_url(path)
        for i in range(self._restCallRequestCounter):
            try:
                with async_timeout.timeout(self._restCallTimout, loop=self._loop):
                    result = await self._websession.post(
                        path, data=body, headers=self.headers
                    )
                    if result.status == 200:
                        if result.content_type == "application/json":
                            ret = await result.json()
                        else:
                            ret = True
                        return ret
                    else:
                        raise HmipWrongHttpStatusError(result.status)
            except (asyncio.TimeoutError, aiohttp.ClientConnectionError):
                # Both exceptions occur when connecting to the server does
                # somehow not work.
                logger.debug("Connection timed out or another error occurred %s" % path)
            except JSONDecodeError as err:
                logger.exception(err)
            finally:
                if result is not None:
                    await result.release()
        raise HmipConnectionError("Failed to connect to HomeMaticIp server") 

def __anext__(self):
        """
            Decode each line using json

        Returns
        -------
        dict
            Decoded JSON data
        """
        if self.response is None:
            logger.info("first connection to the stream")
            await self.connect()
            return {'connected': True}

        line = b''
        try:
            if self.state != NORMAL:
                if self._reconnecting:
                    return await self.restart_stream()
                else:
                    return await self.init_restart()

            while not line:
                if self.response.content.at_eof():
                    logger.debug("Received EOF")
                    self.state = EOF
                    return await self.init_restart()

                with async_timeout.timeout(90):
                    line = await self.response.content.readline()
                    line = line.strip(b'\r\n')
                    logger.debug("received data: %s" % line)

            if line in rate_limit_notices:
                logger.debug("raising StreamLimit")
                raise StreamLimit(line)

            logger.debug("decoding data")
            return self.loads(line)

        except HandledErrors as e:
            logger.debug("handling error %s: %s" % (e.__class__.__name__, e))
            self.state = ERROR
            return await self.init_restart()

        except ClientConnectionError:
            logger.debug("Disconnected from stream")
            self.state = DISCONNECTION
            return await self.init_restart()

        except CancelledError:
            logger.debug("Stopping stream")
            raise

        except Exception as e:
            self.state = ERROR
            return await self.init_restart(error=e) 

def _do_get_response(self, request, operation_model):
        try:
            logger.debug("Sending http request: %s", request)
            history_recorder.record('HTTP_REQUEST', {
                'method': request.method,
                'headers': request.headers,
                'streaming': operation_model.has_streaming_input,
                'url': request.url,
                'body': request.body
            })
            service_id = operation_model.service_model.service_id.hyphenize()
            event_name = 'before-send.%s.%s' % (
                service_id, operation_model.name)
            responses = await self._event_emitter.emit(event_name,
                                                       request=request)
            http_response = first_non_none_response(responses)
            if http_response is None:
                http_response = await self._send(request)
        except aiohttp.ClientConnectionError as e:
            e.request = request  # botocore expects the request property
            return None, e
        except aiohttp.http_exceptions.BadStatusLine:
            better_exception = ConnectionClosedError(
                endpoint_url=request.url, request=request)
            return None, better_exception
        except Exception as e:
            logger.debug("Exception received when sending HTTP request.",
                         exc_info=True)
            return None, e

        # This returns the http_response and the parsed_data.
        response_dict = await convert_to_response_dict(http_response,
                                                       operation_model)

        http_response_record_dict = response_dict.copy()
        http_response_record_dict['streaming'] = \
            operation_model.has_streaming_output
        history_recorder.record('HTTP_RESPONSE', http_response_record_dict)

        protocol = operation_model.metadata['protocol']
        parser = self._response_parser_factory.create_parser(protocol)
        parsed_response = parser.parse(
            response_dict, operation_model.output_shape)
        history_recorder.record('PARSED_RESPONSE', parsed_response)
        return (http_response, parsed_response), None

    # NOTE: The only line changed here changing time.sleep to asyncio.sleep 

def handle_session(self):
        assert self.response is not None, "Response is not specified."

        # session was interrupted
        if self.session.interrupted:
            await self.send(close_frame(1002, "Connection interrupted"))

        # session is closing or closed
        elif self.session.state in (STATE_CLOSING, STATE_CLOSED):
            await self.session._remote_closed()
            await self.send(close_frame(3000, "Go away!"))

        else:
            # acquire session
            try:
                await self.manager.acquire(self.session)
            except SessionIsAcquired:
                await self.send(close_frame(2010, "Another connection still open"))
            else:
                try:
                    while True:
                        if self.timeout:
                            try:
                                frame, text = await asyncio.wait_for(
                                    self.session._wait(), timeout=self.timeout
                                )
                            except asyncio.futures.TimeoutError:
                                frame, text = FRAME_MESSAGE, "a[]"
                        else:
                            frame, text = await self.session._wait()

                        if frame == FRAME_CLOSE:
                            await self.session._remote_closed()
                            await self.send(text)
                            return
                        else:
                            stop = await self.send(text)
                            if stop:
                                break
                except asyncio.CancelledError:
                    await self.session._remote_close(exc=aiohttp.ClientConnectionError)
                    await self.session._remote_closed()
                    raise
                except SessionIsClosed:
                    pass
                finally:
                    await self.manager.release(self.session) 

def listen(self, room_name: str) -> None:
        """Main run loop"""
        try:
            user, repo = self.rooms[room_name].split('/')
            logger.error("Listening in %s for repo %s/%s", room_name, user, repo)
            message = None
            while True:
                try:
                    room = await self._api.get_room(room_name)
                    logger.info("%s: joining %s", self, room_name)
                    await self._api.join_room(self._user, room)
                    logger.info("%s: listening in %s", self, room_name)
                    async for message in self._api.iter_chat(room):
                        # getting a new ghapi object for every message because our
                        # creds time out. Ideally, the api class would take care of that.
                        ghapi = await self._ghappapi.get_github_api(False, user, repo)
                        await self.handle_msg(room, message, ghapi)

                # on timeouts, we just run log into the room again
                except (aiohttp.ClientConnectionError, asyncio.TimeoutError):
                    pass

                # http errors just get logged
                except aiohttp.ClientResponseError as exc:
                    logger.exception("HTTP Error Code %s while listening to room %s",
                                     exc.code, room_name)

                # asyncio cancellation needs to be passed up
                except asyncio.CancelledError: # pylint: disable=try-except-raise
                    raise

                # the rest, we just log so that we remain online after an error
                except Exception:  # pylint: disable=broad-except
                    logger.exception("Unexpected exception caught. Last message: '%s'", message)

                await asyncio.sleep(1)
        except asyncio.CancelledError:
            logger.error("%s: stopped listening in %s", self, room_name)

            # we need a new session here as the one we got passed might have been
            # closed already when we get cancelled
            async with aiohttp.ClientSession() as session:
                self._api._session = session
                await self._api.leave_room(self._user, room)
                logger.error("%s: left room %s", self, room_name) 

def _send(self, payload, processor):
        """Send a payload to be converted to JSON.

        Handles temporary connection issues.  Daemon reponse errors
        are raise through DaemonError.
        """
        def log_error(error):
            nonlocal last_error_log, retry
            now = time.time()
            if now - last_error_log > 60:
                last_error_log = now
                self.logger.error(f'{error}  Retrying occasionally...')
            if retry == self.max_retry and self.failover():
                retry = 0

        on_good_message = None
        last_error_log = 0
        data = json.dumps(payload)
        retry = self.init_retry
        while True:
            try:
                result = await self._send_data(data)
                result = processor(result)
                if on_good_message:
                    self.logger.info(on_good_message)
                return result
            except asyncio.TimeoutError:
                log_error('timeout error.')
            except aiohttp.ServerDisconnectedError:
                log_error('disconnected.')
                on_good_message = 'connection restored'
            except aiohttp.ClientConnectionError:
                log_error('connection problem - is your daemon running?')
                on_good_message = 'connection restored'
            except aiohttp.ClientError as e:
                log_error(f'daemon error: {e}')
                on_good_message = 'running normally'
            except WarmingUpError:
                log_error('starting up checking blocks.')
                on_good_message = 'running normally'
            except WorkQueueFullError:
                log_error('work queue full.')
                on_good_message = 'running normally'

            await asyncio.sleep(retry)
            retry = max(min(self.max_retry, retry * 2), self.init_retry) 

def run(self, dispatcher, tracker, domain):
        json_body = self._action_call_format(tracker, domain)

        if not self.action_endpoint:
            raise Exception("The model predicted the custom action '{}' "
                            "but you didn't configure an endpoint to "
                            "run this custom action. Please take a look at "
                            "the docs and set an endpoint configuration. "
                            "{}/customactions/"
                            "".format(self.name(), DOCS_BASE_URL))

        try:
            logger.debug("Calling action endpoint to run action '{}'."
                         "".format(self.name()))
            response = await self.action_endpoint.request(
                json=json_body, method="post", timeout=DEFAULT_REQUEST_TIMEOUT)
            self._validate_action_result(response)

            events_json = response.get("events", [])
            responses = response.get("responses", [])
            await self._utter_responses(responses, dispatcher, tracker)

            evts = events.deserialise_events(events_json)
            return evts

        except ClientResponseError as e:
            if e.status == 400:
                response_data = json.loads(e.text)
                exception = ActionExecutionRejection(
                    response_data["action_name"],
                    response_data.get("error")
                )
                logger.debug(exception.message)
                raise exception
            else:
                raise Exception("Failed to execute custom action.") from e

        except aiohttp.ClientConnectionError as e:
            logger.error("Failed to run custom action '{}'. Couldn't connect "
                         "to the server at '{}'. Is the server running? "
                         "Error: {}".format(self.name(),
                                            self.action_endpoint.url,
                                            e))
            raise Exception("Failed to execute custom action.")

        except aiohttp.ClientError as e:
            # not all errors have a status attribute, but
            # helpful to log if they got it

            # noinspection PyUnresolvedReferences
            status = getattr(e, 'status', None)
            logger.error("Failed to run custom action '{}'. Action server "
                         "responded with a non 200 status code of {}. "
                         "Make sure your action server properly runs actions "
                         "and returns a 200 once the action is executed. "
                         "Error: {}".format(self.name(), status, e))
            raise Exception("Failed to execute custom action.")