def test_it_logs_messages(self):
        asyncio.set_event_loop_policy(uvloop.EventLoopPolicy())
        loop = asyncio.get_event_loop()

        async def test():
            reader = asyncio.StreamReader(loop=loop)
            protocol = asyncio.StreamReaderProtocol(reader)

            transport, _ = await loop.connect_read_pipe(
                lambda: protocol, self.read_pipe
            )

            logger = Logger.with_default_handlers()
            await logger.info("Xablau")

            logged_content = await reader.readline()
            self.assertEqual(logged_content, b"Xablau\n")

            transport.close()
            await logger.shutdown()

        loop.run_until_complete(test()) 

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 handshake(self):
        if sys.platform == 'linux' or sys.platform == 'darwin':
            self.sock_reader, self.sock_writer = await asyncio.open_unix_connection(self.ipc_path, loop=self.loop)
        elif sys.platform == 'win32' or sys.platform == 'win64':
            self.sock_reader = asyncio.StreamReader(loop=self.loop)
            reader_protocol = asyncio.StreamReaderProtocol(
                self.sock_reader, loop=self.loop)
            try:
                self.sock_writer, _ = await self.loop.create_pipe_connection(lambda: reader_protocol, self.ipc_path)
            except FileNotFoundError:
                raise InvalidPipe
        self.send_data(0, {'v': 1, 'client_id': self.client_id})
        data = await self.sock_reader.read(1024)
        code, length = struct.unpack('<ii', data[:8])
        if self._events_on:
            self.sock_reader.feed_data = self.on_event 

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 stream_from_fd(fd, loop):
    """Recieve a streamer for a given file descriptor."""
    reader = asyncio.StreamReader(loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    waiter = asyncio.futures.Future(loop=loop)

    transport = UnixFileDescriptorTransport(
        loop=loop,
        fileno=fd,
        protocol=protocol,
        waiter=waiter,
    )

    try:
        yield from waiter
    except Exception:
        transport.close()

    if loop.get_debug():
        logger.debug("Read fd %r connected: (%r, %r)", fd, transport, protocol)
    return reader, transport 

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 __init__(self, num_tokens_per_source, input_queue_maxsize, write, read):
        super().__init__(num_tokens_per_source)
        self._input_queue = asyncio.Queue(input_queue_maxsize)
        self._write = write

        loop = asyncio.get_event_loop()
        self._stream_reader = asyncio.StreamReader()
        def protocol_factory():
            return asyncio.StreamReaderProtocol(self._stream_reader)
        pipe = os.fdopen(read, mode='r')
        self._transport, _ = loop.run_until_complete(
            loop.connect_read_pipe(protocol_factory, pipe)) 

def get_producer_wrapper(self):
        async def receiver():
            stream_reader = asyncio.StreamReader()
            def protocol_factory():
                return asyncio.StreamReaderProtocol(stream_reader)
            transport = await asyncio.get_event_loop().connect_read_pipe(
                protocol_factory, os.fdopen(self._read, mode='r'))

            # TODO do we need to close transport when filter gets garbage
            # collected?

            while True:
                size_bytes = await stream_reader.readexactly(
                    _NUM_BYTES_FOR_SIZE)
                size_of_message = int.from_bytes(size_bytes, _BYTEORDER)

                input_frame = gabriel_pb2.InputFrame()
                input_frame.ParseFromString(
                    await stream_reader.readexactly(size_of_message))
                self._latest_input_frame = input_frame
                self._frame_available.set()

        async def producer():
            if not self._started_receiver:
                self._started_receiver = True
                assert os.getpid() == self._constructor_pid
                asyncio.ensure_future(receiver())

            await self._frame_available.wait()

            # Clear because we are sending self._latest_input_frame
            self._frame_available.clear()

            return self._latest_input_frame

        return ProducerWrapper(producer=producer, source_name=self._source_name) 

def get_reader(self):
        if self._reader is None:
            self._reader = asyncio.StreamReader()
            protocol = asyncio.StreamReaderProtocol(self._reader)
            loop = asyncio.get_event_loop()
            await loop.connect_read_pipe(lambda: protocol, sys.stdin)
        return self._reader 

def get_pipe_reader(self, fd_reader):
        reader = asyncio.StreamReader()
        protocol = asyncio.StreamReaderProtocol(reader)
        try:
            await self.loop.connect_read_pipe(lambda: protocol, fd_reader)
        except:
            return None
        return reader 

def get_pipe_reader(self, fd_reader):
        reader = asyncio.StreamReader()
        protocol = asyncio.StreamReaderProtocol(reader)
        try:
            await self.loop.connect_read_pipe(lambda: protocol, fd_reader)
        except:
            return None
        return reader 

def wait_on_fail(func):
    """Test decorator for debugging. It pause test execution on failure and wait
    for user input. It's useful to manually inspect system state just after test
    fails, before executing any cleanup.

    Usage: decorate a test you are debugging.
    DO IT ONLY TEMPORARILY, DO NOT COMMIT!
    """

    @functools.wraps(func)
    def wrapper(self, *args, **kwargs):
        try:
            func(self, *args, **kwargs)
        except:
            print('FAIL\n')
            traceback.print_exc()
            print('Press return to continue:', end='')
            sys.stdout.flush()
            reader = asyncio.StreamReader(loop=self.loop)
            transport, protocol = self.loop.run_until_complete(
                self.loop.connect_read_pipe(
                    lambda: asyncio.StreamReaderProtocol(reader),
                    sys.stdin))
            self.loop.run_until_complete(reader.readline())
            raise

    return wrapper 

def stream_as_generator(loop, stream):
        reader = asyncio.StreamReader(loop=loop)
        reader_protocol = asyncio.StreamReaderProtocol(reader)
        await loop.connect_read_pipe(lambda: reader_protocol, stream)

        while True:
            line = await reader.readline()
            if not line:  # EOF.
                break
            yield line 

def make_read_pipe_stream_reader(
    loop, read_pipe
) -> Tuple[asyncio.StreamReader, asyncio.ReadTransport]:
    reader = asyncio.StreamReader(loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader)

    transport, protocol = await loop.connect_read_pipe(
        lambda: protocol, read_pipe
    )
    return reader, transport 

def create_direct_streams_pair(loop=None, latency=None):
    loop = loop or asyncio.get_event_loop()
    r1 = asyncio.StreamReader(loop=loop)
    r2 = asyncio.StreamReader(loop=loop)

    def protocol_f(reader):
        def f():
            return asyncio.StreamReaderProtocol(reader, loop=loop)
        return f
    (t1, p1), (t2, p2) = create_direct_transport_pair(protocol_f(r1), protocol_f(r2))
    w1 = asyncio.StreamWriter(t1, p1, r1, loop)
    w2 = asyncio.StreamWriter(t2, p2, r2, loop)
    return (r1, w1), (r2, w2) 

def test_read_all_from_pipe_reader(self):
        # See asyncio issue 168.  This test is derived from the example
        # subprocess_attach_read_pipe.py, but we configure the
        # StreamReader's limit so that twice it is less than the size
        # of the data writter.  Also we must explicitly attach a child
        # watcher to the event loop.

        code = """\
import os, sys
fd = int(sys.argv[1])
os.write(fd, b'data')
os.close(fd)
"""
        rfd, wfd = os.pipe()
        args = [sys.executable, '-c', code, str(wfd)]

        pipe = open(rfd, 'rb', 0)
        reader = asyncio.StreamReader(loop=self.loop, limit=1)
        protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop)
        transport, _ = self.loop.run_until_complete(
            self.loop.connect_read_pipe(lambda: protocol, pipe))

        watcher = asyncio.SafeChildWatcher()
        watcher.attach_loop(self.loop)
        try:
            asyncio.set_child_watcher(watcher)
            create = asyncio.create_subprocess_exec(*args,
                                                    pass_fds={wfd},
                                                    loop=self.loop)
            proc = self.loop.run_until_complete(create)
            self.loop.run_until_complete(proc.wait())
        finally:
            asyncio.set_child_watcher(None)

        os.close(wfd)
        data = self.loop.run_until_complete(reader.read(-1))
        self.assertEqual(data, b'data') 

def test_streamreader_constructor(self):
        self.addCleanup(asyncio.set_event_loop, None)
        asyncio.set_event_loop(self.loop)

        # asyncio issue #184: Ensure that StreamReaderProtocol constructor
        # retrieves the current loop if the loop parameter is not set
        reader = asyncio.StreamReader()
        self.assertIs(reader._loop, self.loop) 

def test_streamreaderprotocol_constructor(self):
        self.addCleanup(asyncio.set_event_loop, None)
        asyncio.set_event_loop(self.loop)

        # asyncio issue #184: Ensure that StreamReaderProtocol constructor
        # retrieves the current loop if the loop parameter is not set
        reader = mock.Mock()
        protocol = asyncio.StreamReaderProtocol(reader)
        self.assertIs(protocol._loop, self.loop) 

def _test_pipe(self):
        ADDRESS = r'\\.\pipe\_test_pipe-%s' % os.getpid()

        with self.assertRaises(FileNotFoundError):
            yield from self.loop.create_pipe_connection(
                asyncio.Protocol, ADDRESS)

        [server] = yield from self.loop.start_serving_pipe(
            UpperProto, ADDRESS)
        self.assertIsInstance(server, windows_events.PipeServer)

        clients = []
        for i in range(5):
            stream_reader = asyncio.StreamReader(loop=self.loop)
            protocol = asyncio.StreamReaderProtocol(stream_reader,
                                                    loop=self.loop)
            trans, proto = yield from self.loop.create_pipe_connection(
                lambda: protocol, ADDRESS)
            self.assertIsInstance(trans, asyncio.Transport)
            self.assertEqual(protocol, proto)
            clients.append((stream_reader, trans))

        for i, (r, w) in enumerate(clients):
            w.write('lower-{}\n'.format(i).encode())

        for i, (r, w) in enumerate(clients):
            response = yield from r.readline()
            self.assertEqual(response, 'LOWER-{}\n'.format(i).encode())
            w.close()

        server.close()

        with self.assertRaises(FileNotFoundError):
            yield from self.loop.create_pipe_connection(
                asyncio.Protocol, ADDRESS)

        return 'done' 

def __init__(self, stream_reader, client_connected_cb, loop, name, timeseal):
        asyncio.StreamReaderProtocol.__init__(self, stream_reader, client_connected_cb=client_connected_cb, loop=loop)
        self.name = name
        self.timeseal = timeseal
        self.connected = False
        self.stateinfo = None 

def test_read_all_from_pipe_reader(self):
        # See asyncio issue 168.  This test is derived from the example
        # subprocess_attach_read_pipe.py, but we configure the
        # StreamReader's limit so that twice it is less than the size
        # of the data writter.  Also we must explicitly attach a child
        # watcher to the event loop.

        code = """\
import os, sys
fd = int(sys.argv[1])
os.write(fd, b'data')
os.close(fd)
"""
        rfd, wfd = os.pipe()
        args = [sys.executable, '-c', code, str(wfd)]

        pipe = open(rfd, 'rb', 0)
        reader = asyncio.StreamReader(loop=self.loop, limit=1)
        protocol = asyncio.StreamReaderProtocol(reader, loop=self.loop)
        transport, _ = self.loop.run_until_complete(
            self.loop.connect_read_pipe(lambda: protocol, pipe))

        watcher = asyncio.SafeChildWatcher()
        watcher.attach_loop(self.loop)
        try:
            asyncio.set_child_watcher(watcher)
            create = asyncio.create_subprocess_exec(*args,
                                                    pass_fds={wfd},
                                                    loop=self.loop)
            proc = self.loop.run_until_complete(create)
            self.loop.run_until_complete(proc.wait())
        finally:
            asyncio.set_child_watcher(None)

        os.close(wfd)
        data = self.loop.run_until_complete(reader.read(-1))
        self.assertEqual(data, b'data') 

def test_streamreader_constructor(self):
        self.addCleanup(asyncio.set_event_loop, None)
        asyncio.set_event_loop(self.loop)

        # asyncio issue #184: Ensure that StreamReaderProtocol constructor
        # retrieves the current loop if the loop parameter is not set
        reader = asyncio.StreamReader()
        self.assertIs(reader._loop, self.loop) 

def test_streamreaderprotocol_constructor(self):
        self.addCleanup(asyncio.set_event_loop, None)
        asyncio.set_event_loop(self.loop)

        # asyncio issue #184: Ensure that StreamReaderProtocol constructor
        # retrieves the current loop if the loop parameter is not set
        reader = mock.Mock()
        protocol = asyncio.StreamReaderProtocol(reader)
        self.assertIs(protocol._loop, self.loop) 

def _test_pipe(self):
        ADDRESS = r'\\.\pipe\_test_pipe-%s' % os.getpid()

        with self.assertRaises(FileNotFoundError):
            yield from self.loop.create_pipe_connection(
                asyncio.Protocol, ADDRESS)

        [server] = yield from self.loop.start_serving_pipe(
            UpperProto, ADDRESS)
        self.assertIsInstance(server, windows_events.PipeServer)

        clients = []
        for i in range(5):
            stream_reader = asyncio.StreamReader(loop=self.loop)
            protocol = asyncio.StreamReaderProtocol(stream_reader,
                                                    loop=self.loop)
            trans, proto = yield from self.loop.create_pipe_connection(
                lambda: protocol, ADDRESS)
            self.assertIsInstance(trans, asyncio.Transport)
            self.assertEqual(protocol, proto)
            clients.append((stream_reader, trans))

        for i, (r, w) in enumerate(clients):
            w.write('lower-{}\n'.format(i).encode())

        for i, (r, w) in enumerate(clients):
            response = yield from r.readline()
            self.assertEqual(response, 'LOWER-{}\n'.format(i).encode())
            w.close()

        server.close()

        with self.assertRaises(FileNotFoundError):
            yield from self.loop.create_pipe_connection(
                asyncio.Protocol, ADDRESS)

        return 'done' 

def collect_stats(infile):
    reader = asyncio.StreamReader()
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    transport, _ = await asyncio.get_event_loop() \
        .connect_read_pipe(lambda: reader_protocol, infile)
    count = 0
    dp = None
    stats = None
    while True:
        try:
            line = await reader.readline()
        except ValueError:
            logging.exception('Too large stats object!')
            break
        if line == b'':
            break
        stats = line
        if dp is None:
            try:
                dp = json.loads(line.decode('ascii'))
            except JSONDecodeError:
                break
        count += 1

    transport.close()

    if dp is None or count == 0:
        return {}

    try:
        stats = json.loads(stats.decode('ascii'))
    except JSONDecodeError:
        stats = {}

    return stats 

def wait_for_input(loop, abort_sem):
    """Wait for user input (enter) or abort signal."""
    reader = asyncio.StreamReader(loop=loop)
    reader_protocol = asyncio.StreamReaderProtocol(reader)
    await loop.connect_read_pipe(lambda: reader_protocol, sys.stdin)
    await asyncio.wait(
        [reader.readline(), abort_sem.acquire()], return_when=asyncio.FIRST_COMPLETED
    ) 

def task():
    rfd, wfd = os.pipe()
    args = [sys.executable, '-c', code, str(wfd)]

    pipe = open(rfd, 'rb', 0)
    reader = asyncio.StreamReader(loop=loop)
    protocol = asyncio.StreamReaderProtocol(reader, loop=loop)
    transport, _ = yield from loop.connect_read_pipe(lambda: protocol, pipe)

    proc = yield from asyncio.create_subprocess_exec(*args, pass_fds={wfd})
    yield from proc.wait()

    os.close(wfd)
    data = yield from reader.read()
    print("read = %r" % data.decode()) 

def connect_read_pipe(file):
    loop = asyncio.get_event_loop()
    stream_reader = asyncio.StreamReader(loop=loop)
    def factory():
        return asyncio.StreamReaderProtocol(stream_reader)
    transport, _ = yield from loop.connect_read_pipe(factory, file)
    return stream_reader, transport


#
# Example
#