blackbull.event

blackbull.event

Event-driven dispatcher.

Implements the minimal Pub/Sub dispatcher used by BlackBull.on / BlackBull.intercept. Three delivery modes are supported:

• Interception (intercept): handlers are awaited in registration order; exceptions propagate to the emitter and abort subsequent interceptors.
• Blocking observation (on(..., blocking=True)): handlers are awaited in registration order before emit returns, but their exceptions are caught and logged — they never reach the emitter or abort siblings. This is the "observe but block" mode: use it when a side effect must complete within the event's lifetime (resource cleanup on scope_completed) yet must not be able to break the thing that emitted it.
• Observation (on): handlers are scheduled as independent asyncio.Tasks (fire-and-forget); exceptions are caught and logged and never reach the emitter or other observers.

The two observation modes share isolation (a failing observer is contained); they differ only in whether emit waits for them. Blocking is the right default for cleanup that must finish before the request context is gone; detached is right for telemetry that must not add latency to the hot path.

Event dataclass

An immutable message dispatched through EventDispatcher.

Attributes:

Name	Type	Description
name	str	The event name (e.g. "app_startup").
detail	dict	Arbitrary per-event data. detail is used (rather than payload) to avoid colliding with HTTP/2 and WebSocket protocol terminology already used in the codebase.

EventDispatcher

Minimal Pub/Sub dispatcher with split interception/observation paths.

Interception handlers (intercept) are awaited in registration order; their exceptions propagate to the emitter. Observation handlers (on) are scheduled via asyncio.create_task (fire-and-forget) and their exceptions are caught and logged — they never reach the emitter.

Observer tasks are tracked so they can be drained at shutdown via :meth:aclose. The drain timeout is configured at construction time (shutdown_timeout); any task still running after the timeout is logged at WARNING and cancelled.

aclose() async

Drain pending observer tasks during shutdown.

Waits up to shutdown_timeout seconds (configured at construction) for all in-flight observer tasks to complete. Any tasks still running after the timeout are logged at WARNING and cancelled.

emit(event) async

Dispatch event to all registered handlers.

Delivery order:

1. Interceptors — awaited in registration order; their exceptions propagate (and abort the remaining interceptors).
2. Blocking observers — awaited in registration order; their exceptions are caught and logged (isolated). emit does not return until they finish, so cleanup registered here is guaranteed to complete within the event's lifetime.
3. Detached observers — scheduled as independent tasks (isolated), tracked so they can be drained at shutdown via :meth:aclose.

has_listeners(event_name)

Return True if any interceptor or observer is registered for event_name.

Hot path: callers use this to skip detail-dict / Event construction when no one will receive the event. defaultdict.get returns None for missing keys without inserting an empty list.

intercept(event_name, handler)

Register an interception handler for event_name.

on(event_name, handler, blocking=False)

Register an observation handler for event_name.

With blocking=False (the default) the handler is scheduled as an independent task when the event fires — it never delays the emitter. With blocking=True the handler is awaited in registration order before emit returns, so a side effect (e.g. resource cleanup) is guaranteed to finish within the event's lifetime; its exceptions are still isolated (logged, never propagated).