codekingpro/portable-devtools

1# Urwid main loop code2#    Copyright (C) 2004-2012  Ian Ward3#    Copyright (C) 2008 Walter Mundt4#    Copyright (C) 2009 Andrew Psaltis5#6#    This library is free software; you can redistribute it and/or7#    modify it under the terms of the GNU Lesser General Public8#    License as published by the Free Software Foundation; either9#    version 2.1 of the License, or (at your option) any later version.10#11#    This library is distributed in the hope that it will be useful,12#    but WITHOUT ANY WARRANTY; without even the implied warranty of13#    MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the GNU14#    Lesser General Public License for more details.15#16#    You should have received a copy of the GNU Lesser General Public17#    License along with this library; if not, write to the Free Software18#    Foundation, Inc., 59 Temple Place, Suite 330, Boston, MA  02111-1307  USA19#20# Urwid web site: https://urwid.org/21 22"""Twisted Reactor based urwid EventLoop implementation.23 24Twisted library is required.25"""26 27from __future__ import annotations28 29import functools30import logging31import sys32import typing33 34from twisted.internet.abstract import FileDescriptor35from twisted.internet.error import AlreadyCalled, AlreadyCancelled36 37from .abstract_loop import EventLoop, ExitMainLoop38 39if typing.TYPE_CHECKING:40    from collections.abc import Callable41    from concurrent.futures import Executor, Future42 43    from twisted.internet.base import DelayedCall, ReactorBase44    from typing_extensions import ParamSpec45 46    _Spec = ParamSpec("_Spec")47    _T = typing.TypeVar("_T")48 49__all__ = ("TwistedEventLoop",)50 51 52class _TwistedInputDescriptor(FileDescriptor):53    def __init__(self, reactor: ReactorBase, fd: int, cb: Callable[[], typing.Any]) -> None:54        self._fileno = fd55        self.cb = cb56        super().__init__(reactor)  # ReactorBase implement full API as required in interfaces57 58    def fileno(self) -> int:59        return self._fileno60 61    def doRead(self):62        return self.cb()63 64    def getHost(self):65        raise NotImplementedError("No network operation expected")66 67    def getPeer(self):68        raise NotImplementedError("No network operation expected")69 70    def writeSomeData(self, data: bytes) -> None:71        raise NotImplementedError("Reduced functionality: read-only")72 73 74class TwistedEventLoop(EventLoop):75    """76    Event loop based on Twisted_77    """78 79    _idle_emulation_delay = 1.0 / 256  # a short time (in seconds)80 81    def __init__(self, reactor: ReactorBase | None = None, manage_reactor: bool = True) -> None:82        """83        :param reactor: reactor to use84        :type reactor: :class:`twisted.internet.reactor`.85        :param: manage_reactor: `True` if you want this event loop to run86                                and stop the reactor.87        :type manage_reactor: boolean88 89        .. WARNING::90           Twisted's reactor doesn't like to be stopped and run again.  If you91           need to stop and run your :class:`MainLoop`, consider setting92           ``manage_reactor=False`` and take care of running/stopping the reactor93           at the beginning/ending of your program yourself.94 95           You can also forego using :class:`MainLoop`'s run() entirely, and96           instead call start() and stop() before and after starting the97           reactor.98 99        .. _Twisted: https://twisted.org/100        """101        super().__init__()102        self.logger = logging.getLogger(__name__).getChild(self.__class__.__name__)103        if reactor is None:104            import twisted.internet.reactor105 106            reactor = twisted.internet.reactor107        self.reactor: ReactorBase = reactor108        self._watch_files: dict[int, _TwistedInputDescriptor] = {}109        self._idle_handle: int = 0110        self._twisted_idle_enabled = False111        self._idle_callbacks: dict[int, Callable[[], typing.Any]] = {}112        self._exc: BaseException | None = None113        self.manage_reactor = manage_reactor114        self._enable_twisted_idle()115 116    def run_in_executor(117        self,118        executor: Executor,119        func: Callable[..., _T],120        *args: object,121        **kwargs: object,122    ) -> Future[_T]:123        raise NotImplementedError(124            "Twisted implement it's own ThreadPool executor. Please use native API for call:\n"125            "'threads.deferToThread(Callable[..., Any], *args, **kwargs)'\n"126            "And use 'addCallback' api for callbacks:\n"127            "'threads.deferToThread(Callable[..., T], *args, **kwargs).addCallback(Callable[[T], None])'"128        )129 130    def alarm(self, seconds: float, callback: Callable[[], typing.Any]) -> DelayedCall:131        """132        Call callback() a given time from now.  No parameters are133        passed to callback.134 135        Returns a handle that may be passed to remove_alarm()136 137        seconds -- floating point time to wait before calling callback138        callback -- function to call from event loop139        """140        handle = self.reactor.callLater(seconds, self.handle_exit(callback))141        return handle142 143    def remove_alarm(self, handle: DelayedCall) -> bool:144        """145        Remove an alarm.146 147        Returns True if the alarm exists, False otherwise148        """149        try:150            handle.cancel()151 152        except (AlreadyCancelled, AlreadyCalled):153            return False154 155        return True156 157    def watch_file(self, fd: int, callback: Callable[[], typing.Any]) -> int:158        """159        Call callback() when fd has some data to read.  No parameters160        are passed to callback.161 162        Returns a handle that may be passed to remove_watch_file()163 164        fd -- file descriptor to watch for input165        callback -- function to call when input is available166        """167        ind = _TwistedInputDescriptor(self.reactor, fd, self.handle_exit(callback))168        self._watch_files[fd] = ind169        self.reactor.addReader(ind)170        return fd171 172    def remove_watch_file(self, handle: int) -> bool:173        """174        Remove an input file.175 176        Returns True if the input file exists, False otherwise177        """178        if handle in self._watch_files:179            self.reactor.removeReader(self._watch_files[handle])180            del self._watch_files[handle]181            return True182        return False183 184    def enter_idle(self, callback: Callable[[], typing.Any]) -> int:185        """186        Add a callback for entering idle.187 188        Returns a handle that may be passed to remove_enter_idle()189        """190        self._idle_handle += 1191        self._idle_callbacks[self._idle_handle] = callback192        return self._idle_handle193 194    def _enable_twisted_idle(self) -> None:195        """196        Twisted's reactors don't have an idle or enter-idle callback197        so the best we can do for now is to set a timer event in a very198        short time to approximate an enter-idle callback.199 200        .. WARNING::201           This will perform worse than the other event loops until we can find a202           fix or workaround203        """204        if self._twisted_idle_enabled:205            return206        self.reactor.callLater(207            self._idle_emulation_delay,208            self.handle_exit(self._twisted_idle_callback, enable_idle=False),209        )210        self._twisted_idle_enabled = True211 212    def _twisted_idle_callback(self) -> None:213        for callback in self._idle_callbacks.values():214            callback()215        self._twisted_idle_enabled = False216 217    def remove_enter_idle(self, handle: int) -> bool:218        """219        Remove an idle callback.220 221        Returns True if the handle was removed.222        """223        try:224            del self._idle_callbacks[handle]225        except KeyError:226            return False227        return True228 229    def run(self) -> None:230        """231        Start the event loop.  Exit the loop when any callback raises232        an exception.  If ExitMainLoop is raised, exit cleanly.233        """234        if not self.manage_reactor:235            return236        self.reactor.run()237        if self._exc:238            # An exception caused us to exit, raise it now239            exc = self._exc240            self._exc = None241            raise exc.with_traceback(exc.__traceback__)242 243    def handle_exit(self, f: Callable[_Spec, _T], enable_idle: bool = True) -> Callable[_Spec, _T | None]:244        """245        Decorator that cleanly exits the :class:`TwistedEventLoop` if246        :class:`ExitMainLoop` is thrown inside of the wrapped function. Store the247        exception info if some other exception occurs, it will be reraised after248        the loop quits.249 250        *f* -- function to be wrapped251        """252 253        @functools.wraps(f)254        def wrapper(*args: _Spec.args, **kwargs: _Spec.kwargs) -> _T | None:255            rval = None256            try:257                rval = f(*args, **kwargs)258            except ExitMainLoop:259                if self.manage_reactor:260                    self.reactor.stop()261            except BaseException as exc:262                print(sys.exc_info())263                self._exc = exc264                if self.manage_reactor:265                    self.reactor.crash()266            if enable_idle:267                self._enable_twisted_idle()268            return rval269 270        return wrapper271