trolleybus 0.2.0

Events

An event defines the payload type passed to the listeners and the type of the value they return. Events are never instantiated.

import trolleybus

class UserLoggedIn(trolleybus.Event[str, None]):
    """payload: username, listeners return nothing"""

class ResolveConfig(trolleybus.Event[str, dict]):
    """payload: config key, listeners return the resolved value"""

If no name is given, the class name is used.

Subscribing

bus = trolleybus.EventBus()

# decorator form
@bus.subscribe(UserLoggedIn, priority=60)
def greet(username: str) -> None:
    ...

# direct form
def audit(username: str) -> None:
    ...

bus.subscribe(UserLoggedIn, audit, priority=40)
bus.unsubscribe(UserLoggedIn, audit)

Subscribing the same listener twice only updates its priority.

Emitting

# call all listeners, collect results, propagate listener errors
bus.broadcast(UserLoggedIn, 'kate')

# call all listeners, capture errors as ListenerResult(value, error)
for result in bus.broadcast_nothrow(UserLoggedIn, 'kate'):
    if not result.ok:
        print('listener failed:', result.error)

# call only the highest priority listener (NoListenersError if none)
config = bus.send_one(ResolveConfig, 'database.url')

# call listeners until one returns a non-None value
config = bus.send_any(ResolveConfig, 'database.url')

The bus exposes two lifecycle events: bus.start() emits OnStart and then OnStarted, bus.stop() emits OnExit.

Class-based subscribers

class Metrics(trolleybus.Subscriber):
    @trolleybus.subscribe(UserLoggedIn)
    def on_user_logged_in(self, username: str) -> None:
        ...

metrics = Metrics(bus)
bus.start()  # handlers attach
bus.stop()   # handlers detach; they re-attach on the next start

Emitters

class LoginService(trolleybus.Subscriber, trolleybus.EmitterMixin):
    def login(self, username: str) -> None:
        self.broadcast(UserLoggedIn, username)

A standalone emitter also exists: emitter = trolleybus.Emitter(bus).