EventBusMiddleware

EventBusMiddleware defines the middleware hook contract used by EventBus in both runtimes.

Interface

Python
TypeScript

from typing import Any
from abxbus import BaseEvent, EventBus, EventHandler, EventResult, EventStatus

class EventBusMiddleware:
    async def on_event_change(self, eventbus: EventBus, event: BaseEvent[Any], status: EventStatus) -> None:
        ...

    async def on_event_result_change(
        self,
        eventbus: EventBus,
        event: BaseEvent[Any],
        event_result: EventResult[Any],
        status: EventStatus,
    ) -> None:
        ...

    async def on_bus_handlers_change(self, eventbus: EventBus, handler: EventHandler, registered: bool) -> None:
        ...

import type { BaseEvent, EventBus, EventHandler, EventResult, EventStatus } from 'abxbus'

export interface EventBusMiddleware {
  onEventChange?(eventbus: EventBus, event: BaseEvent, status: EventStatus): void | Promise<void>
  onEventResultChange?(
    eventbus: EventBus,
    event: BaseEvent,
    event_result: EventResult,
    status: EventStatus
  ): void | Promise<void>
  onBusHandlersChange?(eventbus: EventBus, handler: EventHandler, registered: boolean): void | Promise<void>
}

Setup with EventBus

Python
TypeScript

from abxbus import EventBus
from abxbus.middlewares import EventBusMiddleware

class AnalyticsMiddleware(EventBusMiddleware):
    pass

bus = EventBus('AppBus', middlewares=[AnalyticsMiddleware])

import { EventBus, type EventBusMiddleware } from 'abxbus'

class AnalyticsMiddleware implements EventBusMiddleware {}

const bus = new EventBus('AppBus', { middlewares: [AnalyticsMiddleware] })

Lifecycle behavior

on_event_change / onEventChange runs on event lifecycle transitions.
on_event_result_change / onEventResultChange runs on handler-result lifecycle transitions.
on_bus_handlers_change / onBusHandlersChange runs when handlers are added/removed.
Hook status values are only pending, started, and completed.
Handler failures are represented on event_result.status == 'error' and event_result.error when the hook status is completed.

Custom middleware example

Python
TypeScript

from abxbus import EventBus, EventStatus
from abxbus.middlewares import EventBusMiddleware

class AnalyticsMiddleware(EventBusMiddleware):
    async def on_event_result_change(self, eventbus, event, event_result, status):
        if status != EventStatus.COMPLETED:
            return
        if event_result.status == 'error':
            return
        print(event.event_type, event_result.handler_name)

    async def on_bus_handlers_change(self, eventbus, handler, registered):
        action = 'registered' if registered else 'unregistered'
        print(eventbus.label, handler.id, action)

bus = EventBus('AppBus', middlewares=[AnalyticsMiddleware])

import {
  BaseEvent,
  EventBus,
  type EventHandler,
  type EventBusMiddleware,
  type EventResult,
  type EventStatus,
} from 'abxbus'

class AnalyticsMiddleware implements EventBusMiddleware {
  async onEventResultChange(
    eventbus: EventBus,
    event: BaseEvent,
    event_result: EventResult,
    status: EventStatus
  ): Promise<void> {
    if (status !== 'completed') {
      return
    }
    if (event_result.status === 'error') {
      return
    }
    console.log(event.event_type, event_result.handler_name)
  }

  async onBusHandlersChange(eventbus: EventBus, handler: EventHandler, registered: boolean): Promise<void> {
    const action = registered ? 'registered' : 'unregistered'
    console.log(eventbus.label, handler.id, action)
  }
}

const bus = new EventBus('AppBus', { middlewares: [AnalyticsMiddleware] })

Built-in implementations

Python
TypeScript

OtelTracingMiddleware
AutoErrorEventMiddleware
AutoReturnEventMiddleware
AutoHandlerChangeEventMiddleware
WALEventBusMiddleware
LoggerEventBusMiddleware
SQLiteHistoryMirrorMiddleware

Getting Started

Features

API Reference

Concurrency Control

Integrations

Further Reading

EventBusMiddleware

Interface

Setup with EventBus

Lifecycle behavior

Custom middleware example

Built-in implementations

Getting Started

Features

API Reference

Concurrency Control

Integrations

Further Reading

​Interface

​Setup with EventBus

​Lifecycle behavior

​Custom middleware example

​Built-in implementations

Interface

Setup with EventBus

Lifecycle behavior

Custom middleware example

Built-in implementations