Pyjo.Reactor.Asyncio - Low-level event reactor on asyncio

from Pyjo.Reactor.Asyncio import Asyncio

# Watch if handle becomes readable or writable
reactor = Asyncio()
reactor.io(first, lambda reactor, writable: print(
    'First handle is writable' if writable else 'First handle is readable'
))

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

import Pyjo.Reactor.Asyncio

# Watch if handle becomes readable or writable
reactor = Pyjo.Reactor.Asyncio.new()
reactor.io(first, lambda reactor, writable: print(
    'First handle is writable' if writable else 'First handle is readable'
))

# Change to watching only if handle becomes writable
reactor.watch(first, False, True)

# Watch a file descriptor if it becomes readable
reactor.io(fd, lambda reactor, writable: print(
    'Second handle is writable' if writable else 'Second handle is readable'
)).watch(fd, True, False)

# Add a timer
def timeout(reactor):
    reactor.remove(first)
    reactor.remove(fd)
    print('Timeout!')

reactor.timer(15, timeout)

# Start reactor if necessary
if not reactor.is_running:
    reactor.start()

Pyjo.Reactor.Asyncio is a low-level event reactor based on asyncio. It watches handles and runs timers in an asyncio event loop, so coroutines, tasks and other users of the loop run along with it.

A reactor uses the event loop it is given, or else the loop that is running, as inside asyncio.run(), or a new event loop of its own, and moves its handles and timers to the loop that is running. In a loop that runs already, is_running is true, start() and one_tick() return right away, as asyncio does not nest event loops, and the handles and timers are watched by that loop.

The loop watches handles with add_reader and add_writer, which the default event loop of asyncio on Windows, asyncio.ProactorEventLoop, does not have. There, a loop of its own is an asyncio.SelectorEventLoop, and a loop that the reactor is given or that is running has to be one too, as of asyncio.run(main(), loop_factory=asyncio.SelectorEventLoop). The asyncio.SelectorEventLoop of Windows watches sockets only, not pipes or files, up to 512 of them, with select.select().

In the debug mode of asyncio, as of PYTHONDEVMODE, a loop logs a callback that takes longer than its slow_callback_duration, 0.1 seconds. The PYJO_ASYNCIO_SLOW_CALLBACK environment variable sets it in seconds for a loop that the reactor makes, as on a machine that is busy, where TLS and an application of a server take longer.

PYTHONDEVMODE=1 PYJO_ASYNCIO_SLOW_CALLBACK=5 python myapp.py daemon

The end of a with block closes the event loop that the reactor made.

See also

Pyjoyment, https://mojolicious.org.

Ported from Mojo::Reactor::Poll of Mojolicious 9.49.

class Pyjo.Reactor.Asyncio.Pyjo_Reactor_Asyncio(loop: AbstractEventLoop | None = None)

Pyjo.Reactor.Asyncio inherits all methods from Pyjo.Reactor 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. Note that this method requires an active timer.

awaitable(promise: Any) → Awaitable[None]

future = reactor.awaitable(promise)

Return a future of loop that is done when the promise is fulfilled or rejected, so that await promise works in any task of the loop.

close() → None

reactor.close()

Close the event loop that the reactor made for itself, as the end of a with block does, after the tasks that are left in it are cancelled and have ended, as asyncio.run() does, and as the end of the program does; an event loop that it was given, or that was running, is left to its owner.

event_loop_modules = ('asyncio', 'selectors')

modules = reactor.event_loop_modules

The names of the modules of the event loop that the reactor runs on, as ('trio',), whose code a finalizer of the garbage collector may interrupt. The cleanup of a stream, a server or a user agent that is freed in a frame of such a module, or of a module of a class of the reactor, waits for the next tick of the reactor, so that it does not change the structures that the event loop is changing. Defaults to ().

class Pyjo_Reactor_MyLoop(Pyjo.Reactor.Poll.object):
    event_loop_modules = ('myloop',)

future() → Future[Any]

future = reactor.future()

Return a new asyncio.Future of loop, which Pyjo.Promise.Pyjo_Promise.to_future() turns a promise into.

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.

# 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, also in a loop that runs already, as inside asyncio.run(), which watches its handles and timers.

property loop: AbstractEventLoop

loop = reactor.loop

The asyncio event loop of the reactor: the one it is given, or else the loop that is running, as inside asyncio.run(), or a loop of its own. A reactor that is not given a loop follows the loop that is running, and moves its handles and timers to it, unless it has tasks in another loop, which cannot move. In the child of a fork, it moves to a new loop of its own, as the selector of the loop of the parent is not the child’s.

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.

one_tick() → None

reactor.one_tick()

Run reactor until an event occurs or no events are being watched anymore.

# 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.

remove(remove: Handle | str) → bool

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

Remove handle or timer.

reset() → None

reactor.reset()

Remove all handles and timers.

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

reactor.run_coroutine(coro, resolve, reject)

Run a coroutine as a task of loop, where it can await anything of the loop, and call resolve with what it returns or reject with the exception it raises. A task that is not done is something to watch, so that start() does not stop before it, and each step of the coroutine is an event, which ends one_tick().

start() → None

reactor.start()

Start watching for I/O and timer events, this will block until stop() is called or no events are being watched anymore.

# 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.

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.

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. 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)