Pyjo.Reactor - Low-level event reactor base class

from Pyjo.Reactor import Reactor

class MyEventLoop(Reactor):
    def again(self, id, after=None): ...
    def io(self, handle, cb): ...
    @property
    def is_running(self): ...
    def next_tick(self, cb): ...
    def one_tick(self): ...
    def recurring(self, after, cb): ...
    def remove(self, remove): ...
    def reset(self): ...
    def start(self): ...
    def stop(self): ...
    def timer(self, after, cb): ...
    def watch(self, handle, read, write): ...

# Or import the module, as in Google Python Style Guide

import Pyjo.Reactor

class MyEventLoop(Pyjo.Reactor.object):
    def again(self, id, after=None): ...
    def io(self, handle, cb): ...
    @property
    def is_running(self): ...
    def next_tick(self, cb): ...
    def one_tick(self): ...
    def recurring(self, after, cb): ...
    def remove(self, remove): ...
    def reset(self): ...
    def start(self): ...
    def stop(self): ...
    def timer(self, after, cb): ...
    def watch(self, handle, read, write): ...

Pyjo.Reactor is an abstract base class for low-level event reactors, like Pyjo.Reactor.Asyncio.

Events

Pyjo.Reactor inherits all events from Pyjo.EventEmitter and can emit the following new ones.

error

reactor.on('error', lambda reactor, err: ...)

Emitted for exceptions caught in callbacks, fatal if unhandled. Note that if this event is unhandled or fails it might kill your program, so you need to be careful.

reactor.on('error', lambda reactor, err: print(f'Something very bad happened: {err}'))

Alternative reactors

Pyjo.Reactor is a namespace that other distributions can add reactors to, and the PYJO_REACTOR environment variable selects one for Pyjo.IOLoop. A reactor on another implementation of the asyncio event loop, like uvloop, is a subclass of Pyjo.Reactor.Asyncio, and one on an event loop that is not asyncio’s, like the ones of Trio or gevent, implements the methods of Pyjo.Reactor.

A reactor is a module named Pyjo.Reactor.<Name>, with a function new that constructs it and the class as object, as the modules of Pyjoyment have them.

# Pyjo/Reactor/MyLoop.py of another distribution
import Pyjo.Reactor.Asyncio

import asyncio
import selectors

class Pyjo_Reactor_MyLoop(Pyjo.Reactor.Asyncio.object):
    def __init__(self, loop=None):
        super().__init__(loop=loop or asyncio.SelectorEventLoop(selectors.SelectSelector()))

new = Pyjo_Reactor_MyLoop.new
object = Pyjo_Reactor_MyLoop

The distribution contains the module and no __init__.py of Pyjo or Pyjo.Reactor: those of Pyjoyment have the base class and keep the namespace open with pkgutil.extend_path(), and another one would replace them when both are installed in the same directory. Its package depends on Pyjoyment and holds only the files of its own reactor:

Pyjo/
    Reactor/
        MyLoop.py

The reactor is found when the distribution is installed next to Pyjoyment or in another directory of sys.path, and the environment variable selects it:

PYJO_REACTOR=Pyjo.Reactor.MyLoop

See also

Pyjoyment, https://mojolicious.org.

Ported from Mojo::Reactor of Mojolicious 9.49.

class Pyjo.Reactor.Future(*args, **kwargs)

class Pyjo.Reactor.HasFileno(*args, **kwargs)

class Pyjo.Reactor.Pyjo_Reactor

Pyjo.Reactor inherits all methods from Pyjo.EventEmitter and implements the following new ones.

again(id: str, after: float | None = None) → None

reactor.again(id)
reactor.again(id, 0.5)

Restart timer and optionally change the invocation time. Meant to be overloaded in a subclass. Note that this method requires an active timer.

awaitable(promise: Any) → Awaitable[None]

awaitable = reactor.awaitable(promise)

Return what await promise waits on, which completes when the promise is fulfilled or rejected. The base class gives one that run_coroutine() drives, without asyncio, and a reactor of an event loop gives one of its loop, so that a promise can be awaited in the tasks of that loop.

classmethod detect() → str

name = Pyjo.Reactor.detect()

Detect and load the best reactor implementation available, will try the value of the PYJO_REACTOR environment variable or Pyjo.Reactor.Asyncio, and return the name of its module.

# Instantiate best reactor implementation available
reactor = sys.modules[Pyjo.Reactor.detect()].new()

future() → Future[Any]

future = reactor.future()

Return a new future, which Pyjo.Promise.Pyjo_Promise.to_future() turns a promise into. The base class gives a concurrent.futures.Future, which needs no event loop and can be waited for in another thread, and a reactor of an event loop gives one of its loop, so that it can be awaited in the tasks of that loop.

io(handle: Handle, cb: IOCallback) → Self

reactor = reactor.io(handle, cb)

Watch handle for I/O events, invoking the callback whenever handle becomes readable or writable. Meant to be overloaded in a subclass.

# Callback will be executed twice if handle becomes readable and writable
reactor.io(handle, lambda reactor, writable: print('Handle is writable' if writable else 'Handle is readable'))

property is_running: bool

boolean = reactor.is_running

Check if reactor is running. Meant to be overloaded in a subclass.

next_tick(cb: TimerCallback) → None

none = reactor.next_tick(cb)

Execute callback as soon as possible, but not before returning or other callbacks that have been registered with this method, always returns None. Meant to be overloaded in a subclass.

one_tick() → None

reactor.one_tick()

Run reactor until an event occurs. Note that this method can recurse back into the reactor, so you need to be careful. Meant to be overloaded in a subclass.

# Don't block longer than 0.5 seconds
id = reactor.timer(0.5, lambda reactor: None)
reactor.one_tick()
reactor.remove(id)

recurring(after: float, cb: TimerCallback) → str

id = reactor.recurring(0.25, cb)

Create a new recurring timer, invoking the callback repeatedly after a given amount of time in seconds. Meant to be overloaded in a subclass.

remove(remove: Handle | str) → bool

boolean = reactor.remove(handle)
boolean = reactor.remove(id)

Remove handle or timer. Meant to be overloaded in a subclass.

reset() → None

reactor.reset()

Remove all handles and timers. Meant to be overloaded in a subclass.

run_coroutine(coro: Coroutine[Any, Any, Any], resolve: Callable[[Any], Anything], reject: Callable[[Any], Anything]) → None

reactor.run_coroutine(coro, resolve, reject)

Run a coroutine and call resolve with what it returns or reject with the exception it raises. The base class runs it at once, until it waits for a promise, as Mojo runs an async sub, and continues it when the promise is fulfilled or rejected, so that it can await the promises of Pyjoyment without asyncio. A reactor of an event loop runs it as a task of its loop, where it can await anything of that loop.

start() → None

reactor.start()

Start watching for I/O and timer events, this will block until stop() is called. Note that some reactors stop automatically if there are no events being watched anymore. Meant to be overloaded in a subclass.

# Start reactor only if it is not running already
if not reactor.is_running:
    reactor.start()

stop() → None

reactor.stop()

Stop watching for I/O and timer events. Meant to be overloaded in a subclass.

timer(after: float, cb: TimerCallback) → str

id = reactor.timer(0.5, cb)

Create a new timer, invoking the callback after a given amount of time in seconds. Meant to be overloaded in a subclass.

watch(handle: Handle, read: bool, write: bool) → Self

reactor = reactor.watch(handle, readable, writable)

Change I/O events to watch handle for with true and false values. Meant to be overloaded in a subclass. Note that this method requires an active I/O watcher.

# Watch only for readable events
reactor.watch(handle, True, False)

# Watch only for writable events
reactor.watch(handle, False, True)

# Watch for readable and writable events
reactor.watch(handle, True, True)

# Pause watching for events
reactor.watch(handle, False, False)

Pyjo.Reactor.detect() → str

name = Pyjo.Reactor.detect()

Detect and load the best reactor implementation available, will try the value of the PYJO_REACTOR environment variable or Pyjo.Reactor.Asyncio, and return the name of its module.

# Instantiate best reactor implementation available
reactor = sys.modules[Pyjo.Reactor.detect()].new()