def handshake(
    remote: kademlia.Node, privkey: datatypes.PrivateKey, token: CancelToken
) -> Tuple[
    bytes, bytes, BasePreImage, BasePreImage, asyncio.StreamReader, asyncio.StreamWriter
]:  # noqa: E501
    """
    Perform the auth handshake with given remote.

    Returns the established secrets and the StreamReader/StreamWriter pair already connected to
    the remote.
    """
    use_eip8 = False
    initiator = HandshakeInitiator(remote, privkey, use_eip8, token)
    reader, writer = await initiator.connect()
    opened_connections[remote.__repr__()] = (reader, writer)
    aes_secret, mac_secret, egress_mac, ingress_mac = await _handshake(
        initiator, reader, writer, token
    )
    return aes_secret, mac_secret, egress_mac, ingress_mac, reader, writer 

def aclosing_multiple_writers(*writers: asyncio.StreamWriter):
    """Closes StreamWriters on clean exits, aborts them on exceptions.

    The "as" clause returns a set, and more StreamWriters can be added to the
    set.
    """
    writers = set(writers)
    try:
        yield writers
    except:
        for w in writers:
            w.transport.abort()
        raise
    else:
        for w in writers:
            w.close()
    finally:
        close_tasks, _ = await asyncio.wait([w.wait_closed() for w in writers])
        for t in close_tasks:
            if t.exception():
                logger.debug(
                    'wait_closed() raised exception: %r', t.exception()) 

def handle_ext_server_connection(
        upstream_host: str,
        upstream_port: int,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
        info: adapters.ExtOrPortClientConnection,
) -> None:
    handler_logger.info('Connection received from %r', info)
    async with contexts.log_unhandled_exc(handler_logger), \
               contexts.aclosing_multiple_writers(writer) as writers:
        try:
            ureader, uwriter = await asyncio.open_connection(
                upstream_host, upstream_port)
        except OSError as e:
            handler_logger.warning(
                'Error while connecting to upstream: %r', e)
            return
        writers.add(writer)
        try:
            await relays.relay(reader, writer, ureader, uwriter)
        except OSError as e:
            handler_logger.warning('Connection from %r caught %r', info, e) 

def __init__(
        self,
        app: ASGIFramework,
        loop: asyncio.AbstractEventLoop,
        config: Config,
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
    ) -> None:
        self.app = app
        self.config = config
        self.loop = loop
        self.protocol: ProtocolWrapper
        self.reader = reader
        self.writer = writer
        self.send_lock = asyncio.Lock()
        self.timeout_lock = asyncio.Lock()

        self._keep_alive_timeout_handle: Optional[asyncio.Task] = None 

def __init__(
        self,
        stream=None,
        level: Union[str, int, LogLevel] = LogLevel.NOTSET,
        formatter: Formatter = None,
        filter: Filter = None,
        *,
        loop: Optional[AbstractEventLoop] = None,
    ) -> None:
        super().__init__(loop=loop)
        if stream is None:
            stream = sys.stderr
        self.stream = stream
        self.level = level
        if formatter is None:
            formatter = Formatter()
        self.formatter: Formatter = formatter
        if filter:
            self.add_filter(filter)
        self.protocol_class = AiologgerProtocol
        self._initialization_lock = asyncio.Lock(loop=self.loop)
        self.writer: Optional[StreamWriter] = None 

def _init_writer(self) -> StreamWriter:
        async with self._initialization_lock:
            if self.writer is not None:
                return self.writer

            transport, protocol = await self.loop.connect_write_pipe(
                self.protocol_class, self.stream
            )

            self.writer = StreamWriter(  # type: ignore # https://github.com/python/typeshed/pull/2719
                transport=transport,
                protocol=protocol,
                reader=None,
                loop=self.loop,
            )
            return self.writer 

def test_init_writer_initializes_a_nonblocking_pipe_streamwriter(
        self
    ):

        handler = AsyncStreamHandler(
            stream=self.write_pipe, level=10, formatter=Mock()
        )

        self.assertFalse(handler.initialized)

        await handler._init_writer()

        self.assertIsInstance(handler.writer, asyncio.StreamWriter)
        self.assertIsInstance(handler.writer._protocol, AiologgerProtocol)
        self.assertEqual(handler.writer.transport._pipe, self.write_pipe)
        self.assertTrue(handler.initialized)

        await handler.close() 

def test_emit_writes_records_into_the_stream(self):
        msg = self.record.msg
        formatter = Mock(format=Mock(return_value=msg))
        writer = Mock(write=Mock(), drain=CoroutineMock())

        with patch(
            "aiologger.handlers.streams.StreamWriter", return_value=writer
        ):
            handler = AsyncStreamHandler(
                level=10, stream=self.write_pipe, formatter=formatter
            )

            await handler.emit(self.record)

            writer.write.assert_called_once_with(
                (msg + handler.terminator).encode()
            )
            writer.drain.assert_awaited_once()
            await handler.close() 

def test_emit_calls_handle_error_if_an_error_occurs(self):
        writer = Mock(write=CoroutineMock(), drain=CoroutineMock())
        with patch(
            "aiologger.handlers.streams.StreamWriter", return_value=writer
        ):
            exc = Exception("XABLAU")
            handler = AsyncStreamHandler(
                level=10,
                stream=self.write_pipe,
                formatter=Mock(format=Mock(side_effect=exc)),
            )
            with asynctest.patch.object(
                handler, "handle_error"
            ) as handle_error:
                await handler.emit(self.record)

                handle_error.assert_awaited_once_with(self.record, exc)
                writer.write.assert_not_awaited()
                writer.drain.assert_not_awaited() 

def create_pipe_streams_pair():
    path = tempfile.mktemp()
    loop = asyncio.get_event_loop()
    server_side = asyncio.Future()

    def factory():

        def client_connected_cb(reader, writer):
            server_side.set_result((reader, writer))
        reader = asyncio.StreamReader(loop=loop)
        return asyncio.StreamReaderProtocol(reader, client_connected_cb, loop=loop)

    server = yield from loop.create_unix_server(factory, path)

    r1 = asyncio.StreamReader(loop=loop)
    protocol = asyncio.StreamReaderProtocol(r1, loop=loop)
    transport, _ = yield from loop.create_unix_connection(
        lambda: protocol, path)
    w1 = asyncio.StreamWriter(transport, protocol, r1, loop)

    r2, w2 = yield from server_side
    server.close()
    return (r1, w1), (r2, w2) 

def __call__(self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:
        """Main entry point for an interpreter session with a single client."""
        self.reader = reader
        self.writer = writer
        self.running = True

        if self.banner:
            writer.write(self.banner)
            await writer.drain()

        while self.running:
            try:
                await self.handle_one_command()
            except ConnectionResetError:
                writer.close()
                self.running = False
                break
            except Exception:
                log.exception("Exception in manhole REPL")
                self.writer.write(traceback.format_exc())
                await self.writer.drain() 

def test_socks5_http_create_connection(event_loop):
    reader = asyncio.StreamReader(loop=event_loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=event_loop)

    transport, _ = await create_connection(
        proxy_url=SOCKS5_IPV4_URL,
        protocol_factory=lambda: protocol,
        host=HTTP_TEST_HOST,
        port=HTTP_TEST_PORT,
    )

    writer = asyncio.StreamWriter(transport, protocol, reader, event_loop)

    request = ("GET /ip HTTP/1.1\r\n"
               "Host: %s\r\n"
               "Connection: close\r\n\r\n" % HTTP_TEST_HOST)

    writer.write(request.encode())
    response = await reader.read(-1)
    assert b'200 OK' in response 

def receive_handshake(
            self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter) -> None:

        try:
            try:
                await self._receive_handshake(reader, writer)
            except BaseException:
                if not reader.at_eof():
                    reader.feed_eof()
                writer.close()
                raise
        except COMMON_RECEIVE_HANDSHAKE_EXCEPTIONS as e:
            peername = writer.get_extra_info("peername")
            self.logger.debug("Could not complete handshake with %s: %s", peername, e)
        except asyncio.CancelledError:
            # This exception should just bubble.
            raise
        except Exception:
            peername = writer.get_extra_info("peername")
            self.logger.exception("Unexpected error handling handshake with %s", peername) 

def get_directly_connected_streams(alice_extra_info: Dict[str, Any] = None,
                                   bob_extra_info: Dict[str, Any] = None,
                                   loop: asyncio.AbstractEventLoop = None) -> TConnectedStreams:
    if loop is None:
        loop = asyncio.get_event_loop()

    alice_reader = asyncio.StreamReader()
    bob_reader = asyncio.StreamReader()

    alice_transport = MemoryWriteTransport(bob_reader, extra=alice_extra_info)
    bob_transport = MemoryWriteTransport(alice_reader, extra=bob_extra_info)

    alice_protocol = MemoryProtocol()
    bob_protocol = MemoryProtocol()

    # Link the alice's writer to the bob's reader, and the bob's writer to the
    # alice's reader.
    bob_writer = asyncio.StreamWriter(bob_transport, bob_protocol, alice_reader, loop=loop)
    alice_writer = asyncio.StreamWriter(alice_transport, alice_protocol, bob_reader, loop=loop)
    return (
        (alice_reader, alice_writer),
        (bob_reader, bob_writer),
    ) 

def open_serial_connection(*,
                           loop=None,
                           limit=asyncio.streams._DEFAULT_LIMIT,
                           **kwargs):
    """A wrapper for create_serial_connection() returning a (reader,
    writer) pair.

    The reader returned is a StreamReader instance; the writer is a
    StreamWriter instance.

    The arguments are all the usual arguments to Serial(). Additional
    optional keyword arguments are loop (to set the event loop instance
    to use) and limit (to set the buffer limit passed to the
    StreamReader.

    This function is a coroutine.
    """
    if loop is None:
        loop = asyncio.get_event_loop()
    reader = asyncio.StreamReader(limit=limit, loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    transport, _ = yield from create_serial_connection(
        loop=loop,
        protocol_factory=lambda: protocol,
        **kwargs)
    writer = asyncio.StreamWriter(transport, protocol, reader, loop)
    return reader, writer


# - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - - -
# test 

def open_connection(host=None, port=None, *,
                          limit, loop=None,
                          parser=None, **kwds):
    # XXX: parser is not used (yet)
    if loop is not None and sys.version_info >= (3, 8):
        warnings.warn("The loop argument is deprecated",
                      DeprecationWarning)
    reader = StreamReader(limit=limit)
    protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await get_event_loop().create_connection(
        lambda: protocol, host, port, **kwds)
    writer = asyncio.StreamWriter(transport, protocol, reader,
                                  loop=get_event_loop())
    return reader, writer 

def open_unix_connection(address, *,
                               limit, loop=None,
                               parser=None, **kwds):
    # XXX: parser is not used (yet)
    if loop is not None and sys.version_info >= (3, 8):
        warnings.warn("The loop argument is deprecated",
                      DeprecationWarning)
    reader = StreamReader(limit=limit)
    protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await get_event_loop().create_unix_connection(
        lambda: protocol, address, **kwds)
    writer = asyncio.StreamWriter(transport, protocol, reader,
                                  loop=get_event_loop())
    return reader, writer 

def _handshake(
    initiator: "HandshakeInitiator",
    reader: asyncio.StreamReader,
    writer: asyncio.StreamWriter,
    token: CancelToken,
) -> Tuple[bytes, bytes, BasePreImage, BasePreImage]:
    """See the handshake() function above.

    This code was factored out into this helper so that we can create Peers with directly
    connected readers/writers for our tests.
    """
    initiator_nonce = keccak(os.urandom(HASH_LEN))
    auth_msg = initiator.create_auth_message(initiator_nonce)
    auth_init = initiator.encrypt_auth_message(auth_msg)
    writer.write(auth_init)

    auth_ack = await token.cancellable_wait(
        reader.read(ENCRYPTED_AUTH_ACK_LEN), timeout=REPLY_TIMEOUT
    )

    if reader.at_eof():
        # This is what happens when Parity nodes have blacklisted us
        # (https://github.com/ethereum/py-evm/issues/901).
        raise HandshakeDisconnectedFailure(
            "%s disconnected before sending auth ack", repr(initiator.remote)
        )

    ephemeral_pubkey, responder_nonce = initiator.decode_auth_ack_message(auth_ack)
    aes_secret, mac_secret, egress_mac, ingress_mac = initiator.derive_secrets(
        initiator_nonce, responder_nonce, ephemeral_pubkey, auth_init, auth_ack
    )

    return aes_secret, mac_secret, egress_mac, ingress_mac 

def connect(self) -> Tuple[asyncio.StreamReader, asyncio.StreamWriter]:
        return await self.cancel_token.cancellable_wait(
            asyncio.open_connection(
                host=self.remote.address.ip, port=self.remote.address.tcp_port
            ),
            timeout=REPLY_TIMEOUT,
        ) 

def receive_handshake(
        self, reader: asyncio.StreamReader, writer: asyncio.StreamWriter
    ) -> None:
        ip, socket, *_ = writer.get_extra_info("peername")
        remote_address = Address(ip, socket)
        if self.peer_pool.chk_dialin_blacklist(remote_address):
            Logger.info_every_n(
                "{} has been blacklisted, refusing connection".format(remote_address),
                100,
            )
            reader.feed_eof()
            writer.close()
        expected_exceptions = (
            TimeoutError,
            PeerConnectionLost,
            HandshakeFailure,
            asyncio.IncompleteReadError,
            HandshakeDisconnectedFailure,
        )
        try:
            await self._receive_handshake(reader, writer)
        except expected_exceptions as e:
            self.logger.debug("Could not complete handshake: %s", e)
            Logger.error_every_n("Could not complete handshake: {}".format(e), 100)
            reader.feed_eof()
            writer.close()
        except OperationCancelled:
            self.logger.error("OperationCancelled")
            reader.feed_eof()
            writer.close()
        except Exception as e:
            self.logger.exception("Unexpected error handling handshake")
            reader.feed_eof()
            writer.close() 

def get_directly_connected_streams() -> TConnectedStreams:
    bob_reader = asyncio.StreamReader()
    alice_reader = asyncio.StreamReader()
    # Link the alice's writer to the bob's reader, and the bob's writer to the
    # alice's reader.
    bob_writer = MockStreamWriter(alice_reader.feed_data)
    alice_writer = MockStreamWriter(bob_reader.feed_data)
    return (
        (alice_reader, cast(asyncio.StreamWriter, alice_writer)),
        (bob_reader, cast(asyncio.StreamWriter, bob_writer)),
    ) 

def connect(self):
        loop = self._loop
        self._closed_fut = create_future(loop=loop)
        if self._security_protocol in ["PLAINTEXT", "SASL_PLAINTEXT"]:
            ssl = None
        else:
            assert self._security_protocol in ["SSL", "SASL_SSL"]
            assert self._ssl_context is not None
            ssl = self._ssl_context
        # Create streams same as `open_connection`, but using custom protocol
        reader = asyncio.StreamReader(limit=READER_LIMIT, loop=loop)
        protocol = AIOKafkaProtocol(self._closed_fut, reader, loop=loop)
        transport, _ = await asyncio.wait_for(
            loop.create_connection(
                lambda: protocol, self.host, self.port, ssl=ssl),
            loop=loop, timeout=self._request_timeout)
        writer = asyncio.StreamWriter(transport, protocol, reader, loop)
        self._reader, self._writer, self._protocol = reader, writer, protocol

        # Start reader task.
        self._read_task = self._create_reader_task()

        # Start idle checker
        if self._max_idle_ms is not None:
            self._idle_handle = self._loop.call_soon(
                self._idle_check, weakref.ref(self))

        if self._version_hint and self._version_hint >= (0, 10):
            await self._do_version_lookup()

        if self._security_protocol in ["SASL_SSL", "SASL_PLAINTEXT"]:
            await self._do_sasl_handshake()

        return reader, writer 

def authenticate(
            self,
            reader: asyncio.StreamReader,
            writer: asyncio.StreamWriter,
    ) -> bool:
        """(async) Authenticate a connecting client.

        Returns:
            True if authentication is successful and False otherwise. The
            caller is responsible for closing the connection in case of
            failure.
        """
        writer.write(enums.ExtOrPortAuthTypes.SAFE_COOKIE
                     + enums.ExtOrPortAuthTypes.END_AUTH_TYPES)
        client_auth_type = await reader.readexactly(1)
        if client_auth_type != enums.ExtOrPortAuthTypes.SAFE_COOKIE:
            return False
        client_nonce = await reader.readexactly(self.nonce_len)
        server_nonce = secrets.token_bytes(self.nonce_len)
        server_hash = self.hash(b''.join((
            self.server_hash_header, client_nonce, server_nonce)))
        writer.write(server_hash + server_nonce)
        client_hash = await reader.readexactly(self.hash_len)
        result = hmac.compare_digest(client_hash, self.hash(b''.join((
            self.client_hash_header, client_nonce, server_nonce))))
        writer.write(int(result).to_bytes(1, 'big'))
        return result 

def _write_ext_msg(
            writer: asyncio.StreamWriter,
            command: bytes,
            body: bytes,
    ) -> None:
        assert len(command) == 2
        body_len = len(body).to_bytes(2, 'big')
        writer.write(command + body_len + body)
        await writer.drain() 

def negotiate_socks4_userid(
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
        host: Union[str, ipaddress.IPv4Address, ipaddress.IPv6Address],
        port: int,
        args: Optional[Dict[str, str]],
) -> None:
    try:
        host = ipaddress.IPv4Address(host)
    except ValueError:
        raise ValueError('SOCKS4 only supports IPv4 address')
    if args:
        args_bytes = encode_args(args)
    else:
        args_bytes = b''
    writer.write(b''.join((
        b'\x04',  # ver
        enums.SOCKS4Command.CONNECT,
        port.to_bytes(2, 'big'),
        host.packed,
        args_bytes,
        b'\0',
    )))
    buf = await reader.readexactly(8)
    assert buf[0] == 0, 'Invalid SOCKS4 reply version'
    reply = enums.SOCKS4Reply(buf[1:2])
    if reply is not enums.SOCKS4Reply.GRANTED:
        raise exceptions.PTSOCKS4ConnectError(reply) 

def _relay_data_side(
        reader: asyncio.StreamReader,
        writer: asyncio.StreamWriter,
) -> None:
    """Pass data and EOF from reader to writer."""
    while True:
        buf = await reader.read(BUF_SIZE)
        if not buf:  # EOF
            break
        writer.write(buf)
        await writer.drain()
    writer.write_eof()
    await writer.drain() 

def relay(
        dreader: asyncio.StreamReader,
        dwriter: asyncio.StreamWriter,
        ureader: asyncio.StreamReader,
        uwriter: asyncio.StreamWriter,
) -> None:
    """Pass data/EOF from dreader to uwriter, and ureader to dwriter.

    Both writers are ensured to be closed upon exiting this function.
    """
    _logger.debug(
        'Relaying %r <=> %r', dwriter.get_extra_info('peername'),
        uwriter.get_extra_info('peername'))
    utask = asyncio.create_task(_relay_data_side(dreader, uwriter))
    dtask = asyncio.create_task(_relay_data_side(ureader, dwriter))
    async with contexts.aclosing_multiple_writers(dwriter, uwriter):
        try:
            await asyncio.gather(utask, dtask)
            _logger.debug(
                'Relay %r <=> %r ended normally',
                dwriter.get_extra_info('peername'),
                uwriter.get_extra_info('peername'))
        except:
            dtask.cancel()
            utask.cancel()
            raise
        finally:
            await asyncio.wait({dtask, utask})
            for t in (dtask, utask):
                if t.exception():
                    _logger.debug(
                        'Relay task %r caught exception %r', t, t.exception()) 

def __init__(self) -> None:
        self.host_ = '127.0.0.1'
        """ The local EPMD is always located on the local host. """

        self.port_ = EPMD_DEFAULT_PORT
        """ We expect local EPMD to be available on the default port. """

        self.reader_ = None  # type: [asyncio.StreamReader, None]
        self.writer_ = None  # type: [asyncio.StreamWriter, None]
        self.n_connection_attempts_ = 5 

def __init__(self, reader, writer):
        self.reader: StreamReader = reader
        self.writer: StreamWriter = writer
        self.bot: 'BaseBot' = get_bot() 

def handle_client(reader: StreamReader, writer: StreamWriter):
    # helper function to read the next message from the client
    async def read():
        return (await reader.readline()).decode('utf8').strip()

    try:
        writer.write(b'Connected to the command server!\n')

        # wait for the client to select a valid channel to send messages
        channel_name = ''
        while channel_name not in channels:
            connected_channels = ', '.join(channels)
            writer.write(f'what channel do you want to join?\noptions: {connected_channels}\n'.encode())
            channel_name = await read()

        # client gave us a valid channel, now get the channel object from the cache
        channel: Channel = channels[channel_name]
        writer.write(b'Send `quit` to disconnect')

        # now we just relay messages send
        while True:
            msg = await read()
            if not msg or msg.lower() == 'quit':
                return

            await channel.send_message(msg)
    except ConnectionResetError:
        return