Python asyncio 模块，open_connection() 实例源码

def _get_connection(target,
                          port,
                          ssl,
                          timeout,
                          loop):
    con = asyncio.open_connection(host=target,
                                  port=port,
                                  ssl=ssl)

    try:
        reader, writer = await asyncio.wait_for(con,
                                                int(timeout),
                                                loop=loop)

        return reader, writer
    except (asyncio.TimeoutError, ConnectionRefusedError):
        # If this is reach -> port closed
        return None, None

def client(host, port):
    reader, writer = await asyncio.open_connection(host, port)

    for i in itertools.count():
        writer.write(b'ping\n')
        response = await reader.readline()
        if response == b'pong\n':
            print(i)
        else:
            return

def tcp_echo_client(self, loop):
        reader, writer = yield from asyncio.open_connection(
            '127.0.0.1', 3535,
            # '114.114.114.114', 53,
            loop=loop)

        d = DNSRecord.question("img.alicdn.com")
        q = d.pack()
        b_req = struct.pack(">H", q.__len__()) + q
        writer.write(b_req)

        data = yield from reader.read()
        resp = DNSRecord.parse(data[2:])
        print('Received: %r' % resp)

        print('Close the socket')
        writer.close()

def echo_client():
    #
    # ????:
    #
    reader, writer = yield from asyncio.open_connection('localhost', 8000)

    writer.write(b'Hello, world\n')
    writer.write(b'What a fine day it is.\n')
    writer.write(END)   # ????

    while True:
        line = yield from reader.readline()
        print('\treceived:', line)
        if line == END or not line:
            break

    # ???, ???????
    writer.write(b"")
    # ?????
    writer.close()

def test_bad_request_response():
    app = Sanic('test_bad_request_response')
    lines = []
    async def _request(sanic, loop):
        connect = asyncio.open_connection('127.0.0.1', 42101)
        reader, writer = await connect
        writer.write(b'not http')
        while True:
            line = await reader.readline()
            if not line:
                break
            lines.append(line)
        app.stop()
    app.run(host='127.0.0.1', port=42101, debug=False, after_start=_request)
    assert lines[0] == b'HTTP/1.1 400 Bad Request\r\n'
    assert lines[-1] == b'Error: Bad Request'

def run(self) -> None:
        self.logger.info(f"connecting to {self.server}...")
        reader, writer = await asyncio.open_connection(
            self.server.host, self.server.port, ssl=self.server.ssl, loop=self.loop
        )
        self.writer = writer

        await self.queue.put(NetworkEvent('connected', self))

        try:
            while not reader.at_eof():
                line = await reader.readline()
                line = line.strip()
                self.logger.debug(">", line)
                if line:
                    event = NetworkEvent(NetworkEventName.RAW_LINE, line)
                    await self.queue.put(event)
        except asyncio.CancelledError:
            self.logger.info("Connection.run was cancelled")
        except ConnectionResetError as e:
            self.logger.warning(f"connection was reset; {e}")
        finally:
            self.close()
            await self.queue.put(NetworkEvent('disconnected', None))

def authenticate(self, password=None):
        """Authenticate with the server using the given password"""

        if password is None:
            password = self.password
        password = password.encode(self.encoding)

        self.rd, self.wr = await asyncio.open_connection(
            self.server, self.port, loop=self.loop
        )

        auth_pkt = RconPacket(next(self.pkt_id), SERVERDATA_AUTH, password)
        await self._send_pkt(auth_pkt)

        auth_resp = await self.read_response()

        if auth_resp.pkt_type != SERVERDATA_AUTH_RESPONSE:
            raise RconError('Received invalid auth response packet')

        if auth_resp.pkt_id == -1:
            raise RconAuthError('Bad password')

        self.authenticated = True

def _async_get_json(url, loop):
    url = urllib.parse.urlsplit(url)
    connect = asyncio.open_connection(url.hostname, 443, ssl=True, loop=loop)
    reader, writer = await connect
    query = ('GET {url.path} HTTP/1.0\r\n'
             'Host: {url.hostname}\r\n'
             '\r\n').format(url=url)
    writer.write(query.encode('latin-1'))
    line = b''
    while True:
        current_line = await reader.readline()
        if not current_line:
            break
        line = current_line
    writer.close()
    return json.loads(line.decode('latin1').rstrip())

def open_connection(self):
        try:
            server = await asyncio.start_server(
                self.handle_connection,
                host=self.host,
                port=self.port,
                loop=self.loop,
            )

            try:
                await server.wait_closed()
            finally:
                server.close()
                await server.wait_closed()

        except asyncio.CancelledError:
            raise
        except Exception:
            logger.exception(
                "Unable to start TCP server on %s:%s.",
                self.host,
                self.port,
            )

def request(self, uri, method, data=None, headers=None):
        """ request """
        if headers is None:
            headers = {}
        host, port, path = Http._parse_uri(uri)

        method = "{method} {path} HTTP/1.0\r\n".format(method=method, path=path)
        self._add_user_agent_header()
        self._headers.update(headers)
        request = method.encode() + self.get_headers().encode() + data

        reader, writer = yield from asyncio.open_connection(host, port, loop=self._loop)
        writer.write(request.encode())

        reply = yield from reader.read()
        writer.close()
        return reply

def connect(self):
        # Tell oidentd our 'self.ident' before connecting
        self.writeIdentFile()

        self.reader, self.writer = yield from asyncio.open_connection(self.host, self.port)
        self.write("NICK {0}\r\nUSER {0} 0 * :{0}\r\n".format(self.nickname))

        while not self.stop_event.is_set():
            line = yield from self.reader.readline()
            if not line:
                break
            line = line.decode("utf-8", errors="ignore").rstrip()
            if line:
                yield from self.parse_data(line)

        self.write("QUIT Quit by user.\r\n")
        self.writer.close()

def test_tcp_server_auto_close_connection(event_loop, unused_tcp_port):
    """Connections are closed automatically when sessions finish."""

    host = '127.0.0.1'
    port = unused_tcp_port

    # Intentionally do not close the writer.
    async def noop(reader, writer):
        pass

    async with TCPServer(host, port, noop, loop=event_loop) as server:
        reader, writer = await asyncio.open_connection(
            host=host,
            port=port,
            loop=event_loop,
        )
        try:
            with pytest.raises(asyncio.IncompleteReadError):
                await asyncio.wait_for(
                    reader.readexactly(1), timeout=0.1,
                )
            server.close()
            await server.wait_closed()
        finally:
            writer.close()

def test_bad_request_response():
    app = Sanic('test_bad_request_response')
    lines = []
    @app.listener('after_server_start')
    async def _request(sanic, loop):
        connect = asyncio.open_connection('127.0.0.1', 42101)
        reader, writer = await connect
        writer.write(b'not http')
        while True:
            line = await reader.readline()
            if not line:
                break
            lines.append(line)
        app.stop()
    app.run(host='127.0.0.1', port=42101, debug=False)
    assert lines[0] == b'HTTP/1.1 400 Bad Request\r\n'
    assert lines[-1] == b'Error: Bad Request'

def mock_dripping_response(self, chunks, **kwargs):
        ip = iter(chunks)

        loop = asyncio.get_event_loop()
        rsock, wsock = socket.socketpair()
        resp = FakeTextResponse('', **kwargs)
        resp.content, readtr = await asyncio.open_connection(sock=rsock)

        def send_next():
            try:
                to_send = next(ip)
            except StopIteration:
                wsock.close()
                return
            wsock.send(to_send)
            loop.call_soon(send_next)

        loop.call_soon(send_next)

        return self._cm(resp, readtr)

def remote_sort():
    reader, writer = yield from asyncio.open_connection('127.0.0.1', 2015)
    print("Generating random list...")
    numbers = [random.randrange(10000) for r in range(10000)]
    data = json.dumps(numbers).encode()
    print("List Generated, Sending data")
    writer.write(len(data).to_bytes(8, 'big'))
    writer.write(data)

    print("Waiting for data...")
    data = yield from reader.readexactly(len(data))
    print("Received data")
    sorted_values = json.loads(data.decode())
    print(sorted_values)
    print('\n')
    writer.close()

def connect(self):
        loop = asyncio.get_event_loop()
        while True:
            try:
                self.reader, self.writer = await asyncio.open_connection(self.server.config['masterserver_ip'],
                                                                         self.server.config['masterserver_port'],
                                                                         loop=loop)
                await self.handle_connection()
            except (ConnectionRefusedError, TimeoutError):
                pass
            except (ConnectionResetError, asyncio.IncompleteReadError):
                self.writer = None
                self.reader = None
            finally:
                logger.log_debug("Couldn't connect to the master server, retrying in 30 seconds.")
                await asyncio.sleep(30)

def connect(self):
        print("Connecting to", self.server)
        asyncio.open_connection()
        reader, self.writer = yield from asyncio.open_connection(self.server.host, self.server.port, loop=self.loop)

        # Observe and route incoming message
        self.notifier = StratumNotifier(reader, self.on_notify)
        self.notifier.run()

        yield from self.subscribe()
        yield from self.authorize()

        while True:
            yield from asyncio.sleep(1)

            if self.notifier.task.done():
                # Notifier failed or wanted to stop procesing
                # Let ServerSwitcher catch this and round-robin connection
                raise self.notifier.task.exception() or Exception("StratumNotifier failed, restarting.")

def connect(self):
        for port in ControlServer.PORT_RANGE:
            try:
                self._reader, self._writer = await asyncio.open_connection(host=ControlServer.HOST, port=port)

                message = await self._reader.readexactly(len(ControlServer.HANDSHAKE_MESSAGE))
                if message != ControlServer.HANDSHAKE_MESSAGE:
                    raise RuntimeError('Unknown control server protocol')
            except Exception as e:
                self.close()
                self._reader = None
                self._writer = None
                logger.debug('failed to connect to port %s: %r', port, e)
            else:
                break
        else:
            raise RuntimeError('Failed to connect to a control server')

def create_connection(service, address=('127.0.0.1', 6000), *,
                      protocol_cls=TBinaryProtocol, timeout=None, loop=None, **kw):
    """Create a thrift connection.
    This function is a :ref:`coroutine <coroutine>`.

    Open a connection to the thrift server by address argument.

    :param service: a thrift service object
    :param address: a (host, port) tuple
    :param protocol_cls: protocol type, default is :class:`TBinaryProtocol`
    :param timeout: if specified, would raise `asyncio.TimeoutError` if one rpc call is longer than `timeout`
    :param loop: :class:`Eventloop <asyncio.AbstractEventLoop>` instance, if not specified, default loop is used.
    :param kw: params relaied to asyncio.open_connection()
    :return: newly created :class:`ThriftConnection` instance.
    """
    host, port = address
    reader, writer = yield from asyncio.open_connection(
        host, port, loop=loop, **kw)
    iprotocol = protocol_cls(reader)
    oprotocol = protocol_cls(writer)

    return ThriftConnection(service, iprot=iprotocol, oprot=oprotocol,
                            address=address, loop=loop, timeout=timeout)

def _get_ssl_ready_socket(host, port, *, loop):
    reader, writer = await asyncio.open_connection(host, port, loop=loop)

    tr = writer.transport
    try:
        sock = _get_socket(tr)
        _set_nodelay(sock)

        writer.write(struct.pack('!ll', 8, 80877103))  # SSLRequest message.
        await writer.drain()
        resp = await reader.readexactly(1)

        if resp == b'S':
            return sock.dup()
        else:
            raise ConnectionError(
                'PostgreSQL server at "{}:{}" rejected SSL upgrade'.format(
                    host, port))
    finally:
        tr.close()

def _open_connection(*, loop, addr, params: _ConnectionParameters):
    if isinstance(addr, str):
        r, w = await asyncio.open_unix_connection(addr, loop=loop)
    else:
        if params.ssl:
            sock = await _get_ssl_ready_socket(*addr, loop=loop)

            try:
                r, w = await asyncio.open_connection(
                    sock=sock,
                    loop=loop,
                    ssl=params.ssl,
                    server_hostname=addr[0])
            except Exception:
                sock.close()
                raise

        else:
            r, w = await asyncio.open_connection(*addr, loop=loop)
            _set_nodelay(_get_socket(w.transport))

    return r, w

def authenticate(self, password=None):
        """Authenticate with the server using the given password"""

        if password is None:
            password = self.password
        password = password.encode(self.encoding)

        self.rd, self.wr = await asyncio.open_connection(
            self.server, self.port, loop=self.loop
        )

        auth_pkt = RconPacket(next(self.pkt_id), SERVERDATA_AUTH, password)
        await self._send_pkt(auth_pkt)

        auth_resp = await self.read_response()

        if auth_resp.pkt_type != SERVERDATA_AUTH_RESPONSE:
            raise RconError('Received invalid auth response packet')

        if auth_resp.pkt_id == -1:
            raise RconAuthError('Bad password')

        self.authenticated = True

def main():
    urls = ["http://google.com", "http://app.ustudio.com"]
    tasks = asyncio.wait([fetch(url) for url in urls])
    loop = asyncio.get_event_loop()
    loop.run_until_complete(tasks)

    print("-----------")

    init = asyncio.start_server(handle_client, "127.0.0.1", 8001, loop=loop)
    server = loop.run_until_complete(init)

    client = asyncio.open_connection("127.0.0.1", 8001)
    response = handle_message(client)
    loop.run_until_complete(response)
    server.close()
    loop.run_until_complete(server.wait_closed())
    loop.close()

    print("Finished.")

def test_bad_request_response():
    app = Mach9('test_bad_request_response')
    lines = []

    @app.listener('after_server_start')
    async def _request(mach9, loop):
        connect = asyncio.open_connection('127.0.0.1', 42101)
        reader, writer = await connect
        writer.write(b'not http')
        while True:
            line = await reader.readline()
            if not line:
                break
            lines.append(line)
        app.stop()
    app.run(host='127.0.0.1', port=42101)
    assert lines[0] == b'HTTP/1.1 400 Bad Request\r\n'
    assert lines[-1] == b'Error: Bad Request'

def get_page(host, port, wait=0):
    """Get a "web page" asynchronously.
    """
    reader, writer = await asyncio.open_connection(host, port)
    writer.write(b'\r\n'.join([
        'GET /{} HTTP/1.0'.format(wait).encode(ENCODING),
        b'Host: %b' % host.encode(ENCODING),
        b'Connection: close',
        b'', b''
    ]))
    header = []
    msg_lines = []
    async for raw_line in reader:
        line = raw_line.decode(ENCODING).strip()
        if not line.strip():
            break
        header.append(line)
    encoding = get_encoding(header)
    async for raw_line in reader:
        line = raw_line.decode(encoding).strip()
        msg_lines.append(line)
    writer.close()
    return '\n'.join(msg_lines)

def remote_sort():
    reader, writer = yield from asyncio.open_connection('127.0.0.1', 2015)
    print("Generating random list...")
    numbers = [random.randrange(10000) for r in range(10000)]
    data = json.dumps(numbers).encode()
    print("List Generated, Sending data")
    writer.write(len(data).to_bytes(8, 'big'))
    writer.write(data)

    print("Waiting for data...")
    data = yield from reader.readexactly(len(data))
    print("Received data")
    sorted_values = json.loads(data.decode())
    print(sorted_values)
    print('\n')
    writer.close()

def create_connection(repetitions):
    reader, writer = await asyncio.open_connection(
        host=HOST, port=PORT)

    start_time = float((await reader.readline()))

    writer.write(repetitions.encode() + b'\n')
    await writer.drain()

    async for line in reader:
        # Sleeping a little to emulate processing time and make
        # it easier to add more simultaneous clients
        await asyncio.sleep(1)

        printer(start_time, 'Got line: ', line.decode(),
                end='')

    writer.close()

def _get_response(self, request_data):

        async def client(addr, loop, request_data):
            reader, writer = await asyncio.open_connection(*addr, loop=loop)
            # send a line
            writer.write(request_data)
            # read it back
            response_data = await reader.read()
            writer.close()
            return response_data

        server, addr = self.app.test_server(self.loop)
        response_data = self.loop.run_until_complete(
            asyncio.Task(client(addr, self.loop, request_data),
                         loop=self.loop))
        server.close()
        self.loop.run_until_complete(server.wait_closed())
        return response_data

def info_client(ip, port, loop):
    """ Implement a client for the INFO command. """

    ## Setup the channel
    reader, writer = yield from asyncio.open_connection(
                ip, port, loop=loop)        
    sr = SReader(reader, writer)

    # Send the FULL command
    sr.put("INFO")

    # Part 1. Get the params and the ipub
    (params, ipub) = yield from sr.get()
    (G, g, h, o) = params

    return params, ipub

def issue_client(ip, port, params, ipub, keypair, public_attr, private_attr, loop):
    """ Implements a client for the ISSUE protocol. """

    ## Setup the channel
    reader, writer = yield from asyncio.open_connection(
                ip, port, loop=loop)        
    sr = SReader(reader, writer)

    # Send the FULL command
    sr.put("ISSUE")

    # Part 2. Send the encrypted attributes to server
    user_token = cred_secret_issue_user(params, keypair, private_attr)
    (pub, EGenc, sig_u) = user_token

    sr.put( (user_token, public_attr) )        

    # Part 3. Get the credential back
    cred = yield from sr.get()
    (u, EncE, sig_s) = cred
    mac = cred_secret_issue_user_decrypt(params, keypair, u, EncE, ipub, public_attr, EGenc, sig_s)

    return mac, user_token, cred

def handle_jobs(job_handler, host, port, *, loop):
    """
    Connects to the remote master and continuously receives calls, executes
    them, then returns a response until interrupted.
    """

    try:

        try:
            reader, writer = await asyncio.open_connection(host, port, loop=loop)
        except OSError:
            logging.error("worker could not connect to server")
            return

        while True:

            try:
                call_encoded = await reader.readuntil(b"\n")
            except (asyncio.IncompleteReadError, ConnectionResetError):
                break
            logging.debug("worker got call")
            call_json = call_encoded.decode("utf-8")
            call = json.loads(call_json)

            response = job_handler(call)

            response_json = json.dumps(response) + "\n"
            response_encoded = response_json.encode("utf-8")
            writer.write(response_encoded)
            logging.debug("worker returned response")

    except KeyboardInterrupt:

        pass

def echo_client(server_address, messages):

    log = logging.getLogger('echo_client')

    # The certificate is created with pymotw.com as the hostname,
    # which will not match when the example code runs
    # elsewhere, so disable hostname verification.
    ssl_context = ssl.create_default_context(
        ssl.Purpose.SERVER_AUTH,
    )
    ssl_context.check_hostname = False
    ssl_context.load_verify_locations('pymotw.crt')

    log.debug('connecting to {} port {}'.format(*server_address))
    reader, writer = await asyncio.open_connection(
        *server_address, ssl=ssl_context)

    # This could be writer.writelines() except that
    # would make it harder to show each part of the message
    # being sent.
    for msg in messages:
        writer.write(msg)
        log.debug('sending {!r}'.format(msg))
    # SSL does not support EOF, so send a null byte to indicate
    # the end of the message.
    writer.write(b'\x00')
    await writer.drain()

    log.debug('waiting for response')
    while True:
        data = await reader.read(128)
        if data:
            log.debug('received {!r}'.format(data))
        else:
            log.debug('closing')
            writer.close()
            return

def echo_client(address, messages):

    log = logging.getLogger('echo_client')

    log.debug('connecting to {} port {}'.format(*address))
    reader, writer = await asyncio.open_connection(*address)

    # This could be writer.writelines() except that
    # would make it harder to show each part of the message
    # being sent.
    for msg in messages:
        writer.write(msg)
        log.debug('sending {!r}'.format(msg))
    if writer.can_write_eof():
        writer.write_eof()
    await writer.drain()

    log.debug('waiting for response')
    while True:
        data = await reader.read(128)
        if data:
            log.debug('received {!r}'.format(data))
        else:
            log.debug('closing')
            writer.close()
            return

def wget(host):
    print('wget %s ...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = await connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    await writer.drain()
    while True:
        line = await reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    writer.close()

def wget(host):
    print('wget %s ...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = yield from connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    yield from writer.drain()
    while True:
        line = yield from reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    # Ignore the body, close the socket
    writer.close()

def connect(self):
        self.reader, self.writer = yield from asyncio.open_connection(
            self.host, self.port, ssl=self.ssl)
        peername = self.writer.get_extra_info('peername')
        if peername:
            self.host, self.port = peername[:2]
        else:
            self.log(1, 'NO PEERNAME???', self.host, self.port, self.ssl)
        self.key = self.host, self.port, self.ssl

def activity(self):
        backoff = 0
        while True:
            try:
                self.reader, self.writer = yield from asyncio.open_connection(
                    self.host, self.port, ssl=self.sslctx, loop=self.loop)
            except Exception as exc:
                backoff = min(args.max_backoff, backoff + (backoff//2) + 1)
                logging.info('Error connecting: %r; sleep %s', exc, backoff)
                yield from asyncio.sleep(backoff, loop=self.loop)
                continue
            backoff = 0
            self.next_id = 0
            self.pending = {}
            self. initialized = True
            try:
                while self.todo:
                    payload, waiter = self.todo.pop()
                    if not waiter.done():
                        yield from self.send(payload, waiter)
                while True:
                    resp_id, resp = yield from self.process()
                    if resp_id in self.pending:
                        payload, waiter = self.pending.pop(resp_id)
                        if not waiter.done():
                            waiter.set_result(resp)
            except Exception as exc:
                self.initialized = False
                self.writer.close()
                while self.pending:
                    req_id, pair = self.pending.popitem()
                    payload, waiter = pair
                    if not waiter.done():
                        self.todo.add(pair)
                logging.info('Error processing: %r', exc)

def test_open_connection(self):
        with test_utils.run_test_server() as httpd:
            conn_fut = asyncio.open_connection(*httpd.address,
                                               loop=self.loop)
            self._basetest_open_connection(conn_fut)

def test_open_connection_no_loop_ssl(self):
        with test_utils.run_test_server(use_ssl=True) as httpd:
            conn_fut = asyncio.open_connection(
                *httpd.address,
                ssl=test_utils.dummy_ssl_context(),
                loop=self.loop)

            self._basetest_open_connection_no_loop_ssl(conn_fut)

def connect(self):
        reader, writer = await open_connection(self.host, self.port, loop=self.loop)
        await self._handshake(reader, writer)

        keys = self.crypto.get_box_keys()
        self.crypto.clean()

        self.read_stream, self.write_stream = get_stream_pair(reader, writer, **keys)
        self.writer = writer
        if self._on_connect:
            await self._on_connect()

def wget(host):
    print('wget %s...' % host)
    conn = asyncio.open_connection(host, 80)
    reader, writer = await conn
    header = 'GET / HTTP/1.0\r\nHost:%s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    await writer.drain()
    while True:
        line = await reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    writer.close()

def wget(host):
    print('wget %s...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = yield from connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    yield from writer.drain()
    while True:
        line = yield from reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    # Ignore the body, close the socket
    writer.close()

def connect(self) -> None:
        if self.host.startswith('/'):
            self.reader, self.writer = await asyncio.open_unix_connection(
                path=self.host, loop=self.loop
            )
        else:
            self.reader, self.writer = await asyncio.open_connection(
                host=self.host, port=self.port, loop=self.loop
            )

        sock = self.writer.transport.get_extra_info('socket')
        sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, int(self.options.pool_options.socket_keepalive))

        if self.host.startswith('/'):
            endpoint = self.host
        else:
            endpoint = '{}:{}'.format(self.host, self.port)
        logger.debug('Established connection to {}'.format(endpoint))
        self.read_loop_task = asyncio.ensure_future(self.read_loop(), loop=self.loop)

        ismaster = IsMaster(await self.command(
            'admin', SON([('ismaster', 1)]), ReadPreference.PRIMARY, DEFAULT_CODEC_OPTIONS
        ))

        self.is_mongos = ismaster.server_type == SERVER_TYPE.Mongos
        self.max_wire_version = ismaster.max_wire_version
        if ismaster.max_bson_size:
            self.max_bson_size = ismaster.max_bson_size
        if ismaster.max_message_size:
            self.max_message_size = ismaster.max_message_size
        if ismaster.max_write_batch_size:
            self.max_write_batch_size = ismaster.max_write_batch_size
        self.is_writable = ismaster.is_writable

        self.slave_ok = not self.is_mongos and self.options.read_preference != ReadPreference.PRIMARY

        if self.options.credentials:
            await self._authenticate()

        # Notify waiters that connection has been established
        self.__connected.set()

def create_sandboxes(n):
    parent_sockets = list()
    sandbox_params = list()
    for i in range(n):
        sandbox_dir = mkdtemp(prefix='jd4.sandbox.')
        root_dir = path.join(sandbox_dir, 'root')
        mkdir(root_dir)
        in_dir = path.join(sandbox_dir, 'in')
        mkdir(in_dir)
        out_dir = path.join(sandbox_dir, 'out')
        mkdir(out_dir)
        parent_socket, child_socket = socketpair()
        parent_sockets.append(parent_socket)

        pid = fork()
        if pid == 0:
            for parent_socket in parent_sockets:
                parent_socket.close()
            _handle_child(child_socket, root_dir, in_dir, out_dir)
        child_socket.close()
        sandbox_params.append((pid, sandbox_dir, in_dir, out_dir, parent_socket))

    async def helper(pid, sandbox_dir, in_dir, out_dir, parent_socket):
        reader, writer = await open_connection(sock=parent_socket)
        return Sandbox(pid, sandbox_dir, in_dir, out_dir, reader, writer)

    return gather(*[helper(*sp) for sp in sandbox_params])

def run(self) -> None:
        self._reader, self._writer = await asyncio.open_connection(self.host, self.port)

        self._bg(self._connect())

        while not self._reader.at_eof():

            try:
                msg = await self._get_message()
            except:
                self._log.exception("Error during receiving")
                raise


            if not msg:
                continue

            for ch in self._on_command_handlers:
                args = msg.args[:len(ch.args)]
                if ch.args == args and (not ch.rest or ch.rest == msg.rest):
                    self._log.debug("Calling command handler {} with input {}".format(ch, msg))
                    await ch.handler(msg)

            if not self._connected:
                continue

            if msg.args[0] in (cc.PRIVMSG, cc.NOTICE):
                sender = self._resolve_sender(msg.prefix)
                recipient = self._resolve_recipient(msg.args[1])
                message = Message(sender, recipient, msg.rest, (msg.args[0] == cc.NOTICE))
                await self._handle_on_message(message)
                continue

            # self._log.info("Unhandled command: {} {}".format(command, kwargs))

        self._writer.close()

        self._log.info("Connection closed, exiting")

def wget(host):
    print('wget %s...' % host)
    connect = asyncio.open_connection(host, 80)
    reader, writer = yield from connect
    header = 'GET / HTTP/1.0\r\nHost: %s\r\n\r\n' % host
    writer.write(header.encode('utf-8'))
    yield from writer.drain()
    while True:
        line = yield from reader.readline()
        if line == b'\r\n':
            break
        print('%s header > %s' % (host, line.decode('utf-8').rstrip()))
    # Ignore the body, close the socket
    writer.close()

def connect(self):
        log.debug('[Connection.connect]: {}'.format(self.key))

        reader, writer = yield from asyncio.open_connection(self.ip,
                                                            self.port,
                                                            ssl=self.ssl,
                                                            loop=self.loop)

        self.reader = reader
        self.writer = writer

def ssl_handshake(self, host):
        log.debug('[Connection.ssl_handshake]: {}, {}'.format(self.key, host))
        transport = self.reader._transport
        raw_socket = transport.get_extra_info('socket', default=None)
        # transport.pause_reading()
        self.reader, self.writer = yield from asyncio.open_connection(
            ssl=True, sock=raw_socket, server_hostname=host)

def connect(self):
        self.reader, self.writer = yield from asyncio.open_connection(
            self.host, self.port, ssl=self.ssl)
        peername = self.writer.get_extra_info('peername')
        if peername:
            self.host, self.port = peername[:2]
        else:
            self.log(1, 'NO PEERNAME???', self.host, self.port, self.ssl)
        self.key = self.host, self.port, self.ssl

def _connect(self) -> None:
        """
        Connect to Graphite Server based on Provided Server Address
        """
        try:
            self._reader, self._writer = await asyncio.open_connection(
                self._graphite_server,
                self._graphite_port,
                loop=self.loop)
        except Exception:
            raise AioGraphiteSendException(
                "Unable to connect to the provided server address %s:%s"
                % self._graphite_server_address
                )