def create_session(self, loop):
        conn = None

        if self.proxy and self.proxy_user:
            conn = aiohttp.ProxyConnector(
                loop=loop,
                limit=self.parallel,
                proxy=self.proxy,
                proxy_auth=aiohttp.BasicAuth(self.proxy_user, self.proxy_password)
            )
        elif self.proxy:
            conn = aiohttp.ProxyConnector(loop=loop, limit=self.parallel, proxy=self.proxy)
        else:
            conn = aiohttp.TCPConnector(loop=loop, limit=self.parallel)

        session = aiohttp.ClientSession(connector=conn)
        return session 

def _request(self, callback, method, uri, data=None, headers=None):
        connector = aiohttp.TCPConnector(loop=self._loop,
                                         verify_ssl=self.verify)
        async with aiohttp.ClientSession(connector=connector) as session:
            self._session = session
            resp = await session.request(method=method,
                                         url=uri,
                                         data=data,
                                         headers=headers)
            body = await resp.text(encoding='utf-8')
            content = await resp.read()
            if resp.status == 599:
                raise base.Timeout
            r = base.Response(resp.status, resp.headers, body, content)
            await session.close()
            return callback(r)

    # python prior 3.4.1 does not play nice with __del__ method 

def get_http_client(app):
    """ get http client """
    if "client" in app:
        return app["client"]

    # first time call, create client interface
    # use shared client so that all client requests
    #   will share the same connection pool
    if "loop" not in app:
        raise KeyError("loop not initialized")
    loop = app["loop"]
    max_tcp_connections = int(config.get("max_tcp_connections"))
    log.info(f"Initiating TCPConnector with limit {max_tcp_connections} connections")
    client = ClientSession(loop=loop, connector=TCPConnector(limit_per_host=max_tcp_connections))
    #create the app object
    app['client'] = client
    return client 

def __init__(self, loop, console, verbose=0, statsd=None, **kw):
        connector = kw.pop("connector", None)
        if connector is None:
            connector = TCPConnector(loop=loop, limit=None)
        super(LoggedClientSession, self).__init__(
            loop=loop,
            request_class=LoggedClientRequest,
            response_class=LoggedClientResponse,
            connector=connector,
            **kw
        )
        self.console = console
        self.request_class = LoggedClientRequest
        self.request_class.verbose = verbose
        self.verbose = verbose
        self.request_class.session = self
        self.request_class.response_class = LoggedClientResponse
        self.statsd = statsd
        self.eventer = EventSender(
            console, [StdoutListener(verbose=self.verbose, console=self.console)]
        ) 

def aiohttp_session(*, auth: Optional[Auth] = None, **kwargs: Any) -> ClientSession:
    headers = {'User-Agent': USER_AGENT}
    if auth:
        headers['Authorization'] = auth.encode()

    # setup SSL
    cafile = config.get('ca')
    if not cafile:
        cafile = certifi.where()
    ssl_context = create_default_context(cafile=cafile)
    try:
        connector = TCPConnector(ssl=ssl_context)
    except TypeError:
        connector = TCPConnector(ssl_context=ssl_context)

    return ClientSession(headers=headers, connector=connector, **kwargs) 

def test_connect(proxy,operator,mode=None):
    conn = aiohttp.TCPConnector(verify_ssl=False)
    async with ClientSession(connector=conn) as s:
        try:
            async with s.get(url=TEST_URL,proxy=proxy[2],
                             timeout=10,allow_redirects=False) as resp:
                page = await resp.text()
                if (resp.status != 200 or str(resp.url) != TEST_URL):
                    utils.log(('[INFO]#proxy:{ip} has been dropped\n'
                               '      #Reason:Abnormal url or return Code').format(ip=proxy[1]))
                    operator.del_proxy_with_id(config.free_ipproxy_table,proxy[0])
                    operator.del_proxy_with_id(config.httpbin_table,proxy[0])
                elif mode == 'add':
                    operator.insert_valid_proxy(id=proxy[0])
                else:
                    operator.update_valid_proxy(id=proxy[0])
                   
        except Exception as e:
            utils.log(('[INFO]#proxy:{ip} has been dropped\n'
                       '      #Reason:{msg}').format(ip=proxy[1],msg=str(e)))
            operator.del_proxy_with_id(config.free_ipproxy_table,proxy[0])
            operator.del_proxy_with_id(config.httpbin_table,proxy[0])
        finally:
            operator.commit() 

def async_fetch(cls, urls, descs=None, cb=None, datas=None, fds=None):
        if descs is None:
            descs = []
        if datas is None:
            datas = []
        if fds is None:
            fds = []
        conn = aiohttp.TCPConnector(limit_per_host=cls.CONNECTIONS_PER_HOST)
        async with aiohttp.ClientSession(
                connector=conn,
                headers={'User-Agent': cls.USER_AGENT}
        ) as session:
            coros = [
                asyncio.ensure_future(cls._async_fetch_one(session, url, desc, cb, data, fd))
                for url, desc, data, fd in zip_longest(urls, descs, datas, fds)
            ]
            with tqdm(asyncio.as_completed(coros),
                      total=len(coros),
                      desc="Downloading", unit="files") as t:
                result = [await coro for coro in t]
        return result 

def _async_loop(self, urls):
        """Asynchronous internal method used to request multiple URLs

        Args:
            urls (list): URLs to fetch

        Returns:
            responses (obj): All URL requests' response coroutines

        """
        results = []
        async with aiohttp.ClientSession(
            connector=aiohttp.TCPConnector(ssl=False)
        ) as session:
            for url in urls:
                result = asyncio.ensure_future(self._get_async(url, session))
                results.append(result)
            responses = await asyncio.gather(*results)
        return responses 

def test_connector_factory(es_params, loop):

    class TCPConnector(aiohttp.TCPConnector):
        used = False

        def __init__(self, *args, **kwargs):
            TCPConnector.used = True
            super(TCPConnector, self).__init__(*args, **kwargs)

    tr = Transport(
        endpoints=[{'host': es_params['host']}],
        sniffer_interval=None,
        loop=loop,
        connector_factory=lambda: TCPConnector(loop=loop)
    )
    assert 1 == len(tr._pool.connections)
    assert TCPConnector.used
    tr.close() 

def __awaitable__(self):
        if self._data is None:
            async with aiohttp.ClientSession(connector=aiohttp.TCPConnector(verify_ssl=self.api.verify_ssl)) as session:
                wait_time = self._wait_time()
                if wait_time is None and self.api:
                    try:
                        await self._make_async_request(session)
                    except ServiceUnavailableException:
                        await asyncio.sleep(60)
                        self._wait_time()
                        await self._make_async_request(session)
                else:
                    await asyncio.sleep(wait_time)
                    await self._make_async_request(session)

        return self 

def start(self, host: str = "127.0.0.1", port: int = 8080) -> None:
        connector = None
        self.log.debug(f"Starting appservice web server on {host}:{port}")
        if self.server.startswith("https://") and not self.verify_ssl:
            connector = aiohttp.TCPConnector(verify_ssl=False)
        self._http_session = aiohttp.ClientSession(loop=self.loop, connector=connector)
        self._intent = AppServiceAPI(base_url=self.server, bot_mxid=self.bot_mxid, log=self.log,
                                     token=self.as_token, state_store=self.state_store,
                                     real_user_content_key=self.real_user_content_key,
                                     client_session=self._http_session).bot_intent()
        ssl_ctx = None
        if self.tls_cert and self.tls_key:
            ssl_ctx = ssl.create_default_context(ssl.Purpose.CLIENT_AUTH)
            ssl_ctx.load_cert_chain(self.tls_cert, self.tls_key)
        self.runner = web.AppRunner(self.app)
        await self.runner.setup()
        site = web.TCPSite(self.runner, host, port, ssl_context=ssl_ctx)
        await site.start() 

def _init_session(self):
        _safe = self.settings.SPIDER_COOKIES_UNSAFE_MODE
        path = self.settings.SPIDER_COOKIES_LOAD_PATH
        _c_cookies = self.settings.SPIDER_COOKIES_CUSTOM
        jar = aiohttp.CookieJar(unsafe=_safe)
        if _c_cookies:
            cookies = _c_cookies
        else:
            cookies = None
        self.conn = aiohttp.TCPConnector(limit=self.settings.CONCURRENCY)
        self.session = aiohttp.ClientSession(connector=self.conn,
                                             cookies=cookies,
                                             cookie_jar=jar)
        if path:
            if os.path.exists(path):
                try:
                    self.session.cookie_jar.load(path)
                    if cookies:
                        self.session.cookie_jar.update_cookies(cookies)
                except:
                    return
                self.logger.debug(f'Loaded [{self.name}] cookie jar.') 

def cloudfare_session(loop):
    sessions = []

    async def go(**kwargs):
        fake = FakeCloudfare(**kwargs)
        info = await fake.start()
        resolver = FakeResolver(info, loop=asyncio.get_event_loop())
        connector = aiohttp.TCPConnector(resolver=resolver,
                                         ssl=False)

        session = aiohttp.ClientSession(connector=connector)
        sessions.append(session)
        return session

    yield go

    for s in sessions:
        loop.run_until_complete(s.close()) 

def test_secure_ok(aiohttp_client, aiohttp_server, ssl_ctx):
    async def handler(request):
        return web.Response()

    app = web.Application()
    app.router.add_get('/', handler)
    await _setup(app, Secure())
    srv = await aiohttp_server(app, ssl=ssl_ctx)
    conn = aiohttp.TCPConnector(ssl=False)
    cl = await aiohttp_client(srv, connector=conn)
    resp = await cl.get('/')
    print(resp.request_info.url)
    assert resp.status == 200
    assert resp.headers['X-Frame-Options'] == 'DENY'
    expected = 'max-age=31536000; includeSubDomains'
    assert resp.headers['Strict-Transport-Security'] == expected
    assert resp.headers['X-Content-Type-Options'] == 'nosniff'
    assert resp.headers['X-XSS-Protection'] == '1; mode=block' 

def test_secure_redirect(aiohttp_client, aiohttp_server, ssl_ctx):
    async def handler(request):
        return web.Response()

    app = web.Application()
    app.router.add_get('/', handler)
    secure = Secure()
    await _setup(app, secure)
    http_srv = await aiohttp_server(app)
    https_srv = await aiohttp_server(app, ssl=ssl_ctx)
    secure._redirect_url = https_srv.make_url('/')
    conn = aiohttp.TCPConnector(ssl=False)
    async with aiohttp.ClientSession(connector=conn) as cl:
        url = http_srv.make_url('/')
        resp = await cl.get(url)
        assert resp.status == 200
        assert resp.request_info.url.scheme == 'https' 

def test_no_x_frame(aiohttp_client, aiohttp_server, ssl_ctx):
    async def handler(request):
        return web.Response()

    app = web.Application()
    app.router.add_get('/', handler)
    await _setup(app, Secure(x_frame=None))
    srv = await aiohttp_server(app, ssl=ssl_ctx)
    conn = aiohttp.TCPConnector(ssl=False)
    cl = await aiohttp_client(srv, connector=conn)
    resp = await cl.get('/')
    print(resp.request_info.url)
    assert resp.status == 200
    assert 'X-Frame-Options' not in resp.headers
    expected = 'max-age=31536000; includeSubDomains'
    assert resp.headers['Strict-Transport-Security'] == expected
    assert resp.headers['X-Content-Type-Options'] == 'nosniff'
    assert resp.headers['X-XSS-Protection'] == '1; mode=block' 

def test_no_sts(aiohttp_client, aiohttp_server, ssl_ctx):
    async def handler(request):
        return web.Response()

    app = web.Application()
    app.router.add_get('/', handler)
    await _setup(app, Secure(sts=None))
    srv = await aiohttp_server(app, ssl=ssl_ctx)
    conn = aiohttp.TCPConnector(ssl=False)
    cl = await aiohttp_client(srv, connector=conn)
    resp = await cl.get('/')
    print(resp.request_info.url)
    assert resp.status == 200
    assert resp.headers['X-Frame-Options'] == 'DENY'
    assert 'Strict-Transport-Security' not in resp.headers
    assert resp.headers['X-Content-Type-Options'] == 'nosniff'
    assert resp.headers['X-XSS-Protection'] == '1; mode=block' 

def test_no_xss(aiohttp_client, aiohttp_server, ssl_ctx):
    async def handler(request):
        return web.Response()

    app = web.Application()
    app.router.add_get('/', handler)
    await _setup(app, Secure(xss=None))
    srv = await aiohttp_server(app, ssl=ssl_ctx)
    conn = aiohttp.TCPConnector(ssl=False)
    cl = await aiohttp_client(srv, connector=conn)
    resp = await cl.get('/')
    print(resp.request_info.url)
    assert resp.status == 200
    assert resp.headers['X-Frame-Options'] == 'DENY'
    expected = 'max-age=31536000; includeSubDomains'
    assert resp.headers['Strict-Transport-Security'] == expected
    assert resp.headers['X-Content-Type-Options'] == 'nosniff'
    assert 'X-XSS-Protection' not in resp.headers 

def test_connect_envvar(monkeypatch):
    monkeypatch.setenv("DOCKER_HOST", "unix:///var/run/does-not-exist-docker.sock")
    docker = Docker()
    assert isinstance(docker.connector, aiohttp.connector.UnixConnector)
    assert docker.docker_host == "unix://localhost"
    with pytest.raises(aiodocker.DockerError):
        await docker.containers.list()
    await docker.close()

    monkeypatch.setenv("DOCKER_HOST", "http://localhost:9999")
    docker = Docker()
    assert isinstance(docker.connector, aiohttp.TCPConnector)
    assert docker.docker_host == "http://localhost:9999"
    with pytest.raises(aiodocker.DockerError):
        await docker.containers.list()
    await docker.close() 

def _send_to_external_chat(self, bot, event, config):
        if event.from_bot:
            # don't send my own messages
            return

        event_timestamp = event.timestamp

        conversation_id = event.conv_id
        conversation_text = event.text

        user_full_name = event.user.full_name
        user_id = event.user_id

        url = config["HUBOT_URL"] + conversation_id
        payload = {"from" : str(user_id.chat_id), "message" : conversation_text}
        headers = {'content-type': 'application/json'}

        connector = aiohttp.TCPConnector(verify_ssl=False)
        asyncio.ensure_future(
            aiohttp.request('post', url, data = json.dumps(payload), headers = headers, connector=connector)
        ).add_done_callback(lambda future: future.result()) 

def fetch_json(self, url, headers=None, error_dict=None):
        conn = aiohttp.TCPConnector(
            family=socket.AF_INET,
            verify_ssl=False,
        )
        data = dict()
        async with self.session.get(url, headers=headers) as resp:
            if resp.status == 200:
                data = await resp.json()
            elif error_dict and resp.status in error_dict.keys():
                for k, v in error_dict.items():
                    if resp.status == k:
                        raise v()
            else:
                raise UnknownServerError()

        return data 

def run(self, host):
        tasks = []
        # 默认limit=100，enable_cleanup_closed设置为True防止ssl泄露，ttl_dns_cache调高dns缓存
        conn = aiohttp.TCPConnector(
            limit=LIMIT,
            enable_cleanup_closed=True,
            ttl_dns_cache=100,
            ssl=False,
        )
        timeout = aiohttp.ClientTimeout(total=60, connect=2)
        async with aiohttp.ClientSession(connector=conn, timeout=timeout) as session:
            for url in self.urls:
                task = asyncio.ensure_future(self.scan(host, url, session))
                tasks.append(task)
            # gather方法是所有请求完成后才有输出
            _ = await asyncio.gather(*tasks)
            # for i in asyncio.as_completed(tasks):  # 类似于线程池中的task一样
            #     answer = await i
    
    # 创建启动任务 

def _recreate(self) -> None:
        """Re-create the connector, aiohttp session, the APIClient and the Redis session."""
        # Use asyncio for DNS resolution instead of threads so threads aren't spammed.
        # Doesn't seem to have any state with regards to being closed, so no need to worry?
        self._resolver = aiohttp.AsyncResolver()

        # Its __del__ does send a warning but it doesn't always show up for some reason.
        if self._connector and not self._connector._closed:
            log.warning(
                "The previous connector was not closed; it will remain open and be overwritten"
            )

        if self.redis_session and not self.redis_session.closed:
            log.warning(
                "The previous redis pool was not closed; it will remain open and be overwritten"
            )

        # Create the redis session
        self.loop.create_task(self._create_redis_session())

        # Use AF_INET as its socket family to prevent HTTPS related problems both locally
        # and in production.
        self._connector = aiohttp.TCPConnector(
            resolver=self._resolver,
            family=socket.AF_INET,
        )

        # Client.login() will call HTTPClient.static_login() which will create a session using
        # this connector attribute.
        self.http.connector = self._connector

        # Its __del__ does send a warning but it doesn't always show up for some reason.
        if self.http_session and not self.http_session.closed:
            log.warning(
                "The previous session was not closed; it will remain open and be overwritten"
            )

        self.http_session = aiohttp.ClientSession(connector=self._connector)
        self.api_client.recreate(force=True, connector=self._connector) 

def get_watcher_status(request: web.Request, params: Any) -> web.Response:
    log.info('GET_WATCHER_STATUS ()')
    watcher_info = await get_watcher_info(request, params['agent_id'])
    connector = aiohttp.TCPConnector()
    async with aiohttp.ClientSession(connector=connector) as sess:
        with _timeout(5.0):
            headers = {'X-BackendAI-Watcher-Token': watcher_info['token']}
            async with sess.get(watcher_info['addr'], headers=headers) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return web.json_response(data, status=resp.status)
                else:
                    data = await resp.text()
                    return web.Response(text=data, status=resp.status) 

def watcher_agent_start(request: web.Request, params: Any) -> web.Response:
    log.info('WATCHER_AGENT_START ()')
    watcher_info = await get_watcher_info(request, params['agent_id'])
    connector = aiohttp.TCPConnector()
    async with aiohttp.ClientSession(connector=connector) as sess:
        with _timeout(20.0):
            watcher_url = watcher_info['addr'] / 'agent/start'
            headers = {'X-BackendAI-Watcher-Token': watcher_info['token']}
            async with sess.post(watcher_url, headers=headers) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return web.json_response(data, status=resp.status)
                else:
                    data = await resp.text()
                    return web.Response(text=data, status=resp.status) 

def watcher_agent_stop(request: web.Request, params: Any) -> web.Response:
    log.info('WATCHER_AGENT_STOP ()')
    watcher_info = await get_watcher_info(request, params['agent_id'])
    connector = aiohttp.TCPConnector()
    async with aiohttp.ClientSession(connector=connector) as sess:
        with _timeout(20.0):
            watcher_url = watcher_info['addr'] / 'agent/stop'
            headers = {'X-BackendAI-Watcher-Token': watcher_info['token']}
            async with sess.post(watcher_url, headers=headers) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return web.json_response(data, status=resp.status)
                else:
                    data = await resp.text()
                    return web.Response(text=data, status=resp.status) 

def watcher_agent_restart(request: web.Request, params: Any) -> web.Response:
    log.info('WATCHER_AGENT_RESTART ()')
    watcher_info = await get_watcher_info(request, params['agent_id'])
    connector = aiohttp.TCPConnector()
    async with aiohttp.ClientSession(connector=connector) as sess:
        with _timeout(20.0):
            watcher_url = watcher_info['addr'] / 'agent/restart'
            headers = {'X-BackendAI-Watcher-Token': watcher_info['token']}
            async with sess.post(watcher_url, headers=headers) as resp:
                if resp.status == 200:
                    data = await resp.json()
                    return web.json_response(data, status=resp.status)
                else:
                    data = await resp.text()
                    return web.Response(text=data, status=resp.status) 

def fixture_medias(event_loop):
    if os.environ.get('FORCE_IPV4', False):
        connector = aiohttp.TCPConnector(family=socket.AF_INET)
    else:
        connector = aiohttp.TCPConnector()

    async def download():
        async with aiohttp.ClientSession(loop=event_loop,
                                         connector=connector) as session:
            await asyncio.gather(*[media.download(session=session)
                                   for media in medias.values()])

    event_loop.run_until_complete(download())
    return medias 

def api_access(func):
    async def process(*args, **kwargs):
        async with aiohttp.ClientSession(headers=dict(Authorization='token {}'.format(args[0].key)),
                                         connector=aiohttp.TCPConnector(verify_ssl=False)) as session:
            kwargs['session'] = session
            return await func(*args, **kwargs)
    return process 

def __init__(self, *, auth=None, connector=None, loop=None):
        self._auth = auth or NoAuthProvider()

        if loop is None:
            loop = asyncio.get_event_loop()
        self._loop = loop

        if connector is None:
            self.connector = aiohttp.TCPConnector(force_close=False, loop=loop)
        else:
            self.connector = connector