Method close

python/python3/tornado/iostream.py:567–613

Close this stream. If ``exc_info`` is true, set the ``error`` attribute to the current exception from `sys.exc_info` (or if ``exc_info`` is a tuple, use that instead of `sys.exc_info`).

(
        self,
        exc_info: Union[
            None,
            bool,
            BaseException,
            Tuple[
                "Optional[Type[BaseException]]",
                Optional[BaseException],
                Optional[TracebackType],
            ],
        ] = False,
    )

        self._maybe_add_error_listener()

    def close(
        self,
        exc_info: Union[
            None,
            bool,
            BaseException,
            Tuple[
                "Optional[Type[BaseException]]",
                Optional[BaseException],
                Optional[TracebackType],
            ],
        ] = False,
    ) -> None:
        """Close this stream.

        If ``exc_info`` is true, set the ``error`` attribute to the current
        exception from `sys.exc_info` (or if ``exc_info`` is a tuple,
        use that instead of `sys.exc_info`).
        """
        if not self.closed():
            if exc_info:
                if isinstance(exc_info, tuple):
                    self.error = exc_info[1]
                elif isinstance(exc_info, BaseException):
                    self.error = exc_info
                else:
                    exc_info = sys.exc_info()
                    if any(exc_info):
                        self.error = exc_info[1]
            if self._read_until_close:
                self._read_until_close = False
                self._finish_read(self._read_buffer_size)
            elif self._read_future is not None:
                # resolve reads that are pending and ready to complete
                try:
                    pos = self._find_read_pos()
                except UnsatisfiableReadError:
                    pass
                else:
                    if pos is not None:
                        self._read_from_buffer(pos)
            if self._state is not None:
                self.io_loop.remove_handler(self.fileno())
                self._state = None
            self.close_fd()
            self._closed = True
        self._signal_closed()

    def _signal_closed(self) -> None:
        futures = []  # type: List[Future]