Class PipeIOStream

python/python3/tornado/iostream.py:1620–1670

Pipe-based `IOStream` implementation. The constructor takes an integer file descriptor (such as one returned by `os.pipe`) rather than an open file object. Pipes are generally one-way, so a `PipeIOStream` can be used for reading or writing but not both. ``PipeIOStream`` is onl

class PipeIOStream(BaseIOStream):
    """Pipe-based `IOStream` implementation.

    The constructor takes an integer file descriptor (such as one returned
    by `os.pipe`) rather than an open file object.  Pipes are generally
    one-way, so a `PipeIOStream` can be used for reading or writing but not
    both.

    ``PipeIOStream`` is only available on Unix-based platforms.
    """

    def __init__(self, fd: int, *args: Any, **kwargs: Any) -> None:
        self.fd = fd
        self._fio = io.FileIO(self.fd, "r+")
        if sys.platform == "win32":
            # The form and placement of this assertion is important to mypy.
            # A plain assert statement isn't recognized here. If the assertion
            # were earlier it would worry that the attributes of self aren't
            # set on windows. If it were missing it would complain about
            # the absence of the set_blocking function.
            raise AssertionError("PipeIOStream is not supported on Windows")
        os.set_blocking(fd, False)
        super().__init__(*args, **kwargs)

    def fileno(self) -> int:
        return self.fd

    def close_fd(self) -> None:
        self._fio.close()

    def write_to_fd(self, data: memoryview) -> int:
        try:
            return os.write(self.fd, data)  # type: ignore
        finally:
            # Avoid keeping to data, which can be a memoryview.
            # See https://github.com/tornadoweb/tornado/pull/2008
            del data

    def read_from_fd(self, buf: Union[bytearray, memoryview]) -> Optional[int]:
        try:
            return self._fio.readinto(buf)  # type: ignore
        except (IOError, OSError) as e:
            if errno_from_exception(e) == errno.EBADF:
                # If the writing half of a pipe is closed, select will
                # report it as readable but reads will fail with EBADF.
                self.close(exc_info=e)
                return None
            else:
                raise
        finally:
            del buf


def doctests() -> Any: