\
    A bidirectional pipe of Python objects.

    >>> async def example1():
    ...     a, b = pipe()
    ...     a.send_nowait('foo')
    ...     print(await b.recv())
    >>> asyncio.run(example1())
    foo
    >>> async def example2():
    ...     a, b = pipe()
    ...     await b.send(eof=True)
    ...     await a.recv()
    >>> asyncio.run(example2())
    Traceback (most recent call last):
      ...
    EOFError

def pipe(maxsize=0, *, loop=None) -> Pipe:
    """\
    A bidirectional pipe of Python objects.

    >>> async def example1():
    ...     a, b = pipe()
    ...     a.send_nowait('foo')
    ...     print(await b.recv())
    >>> asyncio.run(example1())
    foo
    >>> async def example2():
    ...     a, b = pipe()
    ...     await b.send(eof=True)
    ...     await a.recv()
    >>> asyncio.run(example2())
    Traceback (most recent call last):
      ...
    EOFError
    """

    class QueueStream:
        def __init__(self, maxsize=0, *, loop=None) -> None:
            self._queue: asyncio.Queue = asyncio.Queue(maxsize, loop=loop)
            self._eof = asyncio.locks.Event(loop=loop)

        def _check_send(self, value: Any, *, eof: bool) -> None:
            if self._eof.is_set():
                raise EOFError("Cannot send after EOF")
            PipeEnd.check_send_args(value, eof=eof)

        def send_nowait(self, value: Any, *, eof: bool) -> None:
            self._check_send(value, eof=eof)

            if eof:
                self._eof.set()
            else:
                self._queue.put_nowait(value)

        async def send(self, value: Any, *, eof: bool) -> None:
            self._check_send(value, eof=eof)

            if eof:
                self._eof.set()
            else:
                await self._queue.put(value)

        async def recv(self) -> Any:
            get = asyncio.create_task(self._queue.get())
            eof = asyncio.create_task(self._eof.wait())

            done, pending = await asyncio.wait([get, eof], return_when=asyncio.FIRST_COMPLETED)

            # cancel get or eof, whichever is not finished
            for task in pending:
                task.cancel()

            if get in done:
                return get.result()
            else:
                raise EOFError

    class _PipeEnd(PipeEnd):
        def __init__(self, send: QueueStream, recv: QueueStream) -> None:
            super().__init__()
            self._send = send
            self._recv = recv

        def send_nowait(self, value: Any=PipeEnd._none, *, eof=False):
            self._send.send_nowait(value, eof=eof)

        async def send(self, value: Any=PipeEnd._none, *, eof=False):
            await self._send.send(value, eof=eof)

        async def recv(self)-> Any:
            return await self._recv.recv()

    a, b = QueueStream(maxsize, loop=loop), QueueStream(maxsize, loop=loop)
    return _PipeEnd(a, b), _PipeEnd(b, a)