olekli/mreventloop

Simple event system for Python building on asyncio and working seamlessly across sockets.

MrEventLoop

Simple event system for Python building on asyncio and working seamlessly across sockets.

Installation

pip install mreventloop

Emitting Events

from mreventloop import emits

@emits('events', [ 'produced' ])
class Producer:
  def requestProduct(self):
    self.events.produced('some product')

You can now connect a callback to the event:

from mreventloop import connect

producer = Producer()
connect(producer, 'produced', lambda product: print(f'{product}'))

You can disconnect all listeners from an event:

from mreventloop import disconnect
disconnect(producer, 'produced')

Event Loop and Slots

Objects can have an event loop building on asyncio:

from mreventloop import has_event_loop, slot

@has_event_loop('event_loop')
class Consumer:
  @slot
  def onProduced(self, product):
    print(f'{product}')

Calling a @slot will not run the corresponding method directly, but queue its execution on the object's event loop. The event loop will run them strictly sequentially in FIFO order. Thus, slots do not need to be reentrant.

The event loop has to be started:

consumer = Consumer()
connect(producer, 'produced', consumer, 'onProduced')
async with consumer.event_loop:
  producer.requestProduct()

After exiting, the event loop will still process all queued events.

When creating an object that has_event_loop, an event loop will automatically be created. But you can also share event loops between objects:

producer1 = Producer()
producer2 = Producer()
consumer1 = Consumer()
consumer2 = Consumer()
consumer2.event_loop = consumer1.event_loop
connect(producer1, 'produced', consumer1, 'onProduced')
connect(producer2, 'produced', consumer2, 'onProduced')
async with consumer1.event_loop:
  producer1.requestProduct()
  producer2.requestProduct()

All calls to slots in both consumer1 and consumer2 will be run sequentially.

Async Slots

Slots can also be coroutines:

import asyncio

@has_event_loop('event_loop')
class ConsumerAsync:
  @slot
  async def onProduced(self, product):
    await asyncio.sleep(0.1)
    print(f'{product}')

The coroutine will be awaited inside the event loop.

Awaiting Events

Events can be awaited:

from mreventloop import SyncEvent

producer = Producer()
sync_event = SyncEvent(producer.events.produced)

async def emit():
  producer.requestProduct()

asyncio.create_task(emit())
result = await sync_event
print(f'{result}')

Events of the Event Loop

The event loop itself emits the following events:

[ 'started', 'stopped', 'active', 'idle', 'exception' ]

Exceptions on the Event Loop

If an exception ocurrs executing a slot, the app will exit. This can be prevented by creating the loop manually with:

from mreventloop import EventLoop

consumer.event_loop = EventLoop(exit_on_exception = False)

In this case, any exception will be emitted through the exception event.

(The general idea here is to just not use exceptions.)

Crossing Sockets

Events and slots across multiple applications can be connected via sockets. This requires one instance of a Broker and any number of Peers. A Peer subscribes to a number of events. If any other Peer publishes such events, they will be emitted by the subscribing Peer.

from mreventloop import Broker, Peer

in_socket_path = 'ipc:///tmp/mreventloop_test_in.sock'
out_socket_path = 'ipc:///tmp/mreventloop_test_out.sock'

producer_peer = Peer(in_socket_path, out_socket_path, [ 'request_product' ], [ 'produced' ])
consumer_peer = Peer(in_socket_path, out_socket_path, [ 'produced' ], [ 'request_product' ])
async with Broker(in_socket_path, out_socket_path), producer_peer, consumer_peer:
  consumer = Consumer()
  connect(consumer, 'request_product', consumer_peer.publish, 'request_product')
  connect(consumer_peer, 'product', consumer, 'onProduct')

  producer = Producer()
  connect(producer, 'produce', producer_peer.publish, 'produce')
  connect(producer_peer, 'request_product', producer, 'onRequestProduct')

  async with producer.event_loop, consumer.event_loop:
    consumer.events.request_product()