INNER CODE UNIT · Python

__init__

mars-tw/anti-gambling-trader-tw · core/ui/server.py:143

    def __init__(self, *args: Any, **kwargs: Any) -> None:
        super().__init__(*args, **kwargs)
        self._owned_lock = threading.RLock()
        self._owned_closing = False
        self._owned_connections: dict[int, Any] = {}
        self._owned_handler_threads: dict[int, threading.Thread] = {}
        self._owned_request_deadlines: dict[int, tuple[int, threading.Timer]] = {}
        self._owned_deadline_generation = 0

    @staticmethod
    def _close_owned_socket(request: Any) -> None:
        try:
            request.shutdown(socket.SHUT_RDWR)
        except (AttributeError, OSError, ValueError):
            pass
        try:
            request.close()
        except (AttributeError, OSError, ValueError):

View source record →