EventBusMiddleware

EventBusMiddleware defines the middleware hook contract used by EventBus in both runtimes. Rust core event dispatch does not expose a public middleware trait yet. Use handler wrappers or inspect EventBus::to_json_value() / event result state directly until Rust middleware parity lands.

Interface

Python
TypeScript
Go
Rust

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>
}

type EventBusMiddleware interface {
	OnEventChange(eventbus *abxbus.EventBus, event *abxbus.BaseEvent, status string)
	OnEventResultChange(eventbus *abxbus.EventBus, event *abxbus.BaseEvent, eventResult *abxbus.EventResult, status string)
	OnBusHandlersChange(eventbus *abxbus.EventBus, handler *abxbus.EventHandler, registered bool)
}

type AnalyticsMiddleware struct {
	abxbus.EventBusMiddlewareBase
}

// Not implemented in the Rust crate yet.
//
// Use explicit handler wrappers, normal Rust context propagation, and
// EventBus::to_json_value() for current Rust observability/custom hooks.

Setup with EventBus

Python
TypeScript
Go
Rust

from abxbus import EventBus
from abxbus.middlewares import EventBusMiddleware

class AnalyticsMiddleware(EventBusMiddleware):
    pass

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

import { EventBus } from 'abxbus'
import type { EventBusMiddleware } from 'abxbus/middlewares'

class AnalyticsMiddleware implements EventBusMiddleware {}

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

type AnalyticsMiddleware struct {
	abxbus.EventBusMiddlewareBase
}

bus := abxbus.NewEventBus("AppBus", &abxbus.EventBusOptions{
	Middlewares: []abxbus.EventBusMiddleware{&AnalyticsMiddleware{}},
})

// Not implemented in the Rust crate yet.

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
Go
Rust

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 EventResult,
  type EventStatus,
} from 'abxbus'
import type { EventBusMiddleware } from 'abxbus/middlewares'

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] })

type AnalyticsMiddleware struct {
	abxbus.EventBusMiddlewareBase
}

func (m *AnalyticsMiddleware) OnEventResultChange(
	eventbus *abxbus.EventBus,
	event *abxbus.BaseEvent,
	eventResult *abxbus.EventResult,
	status string,
) {
	if status != "completed" || eventResult.Status == abxbus.EventResultError {
		return
	}
	fmt.Println(event.EventType, eventResult.HandlerName)
}

func (m *AnalyticsMiddleware) OnBusHandlersChange(eventbus *abxbus.EventBus, handler *abxbus.EventHandler, registered bool) {
	action := "unregistered"
	if registered {
		action = "registered"
	}
	fmt.Println(eventbus.Label(), handler.ID, action)
}

bus := abxbus.NewEventBus("AppBus", &abxbus.EventBusOptions{
	Middlewares: []abxbus.EventBusMiddleware{&AnalyticsMiddleware{}},
})

// Not implemented in the Rust crate yet.

Built-in implementations

Python
TypeScript
Go
Rust

OtelTracingMiddleware
AutoErrorEventMiddleware
AutoReturnEventMiddleware
AutoHandlerChangeEventMiddleware
WALEventBusMiddleware
LoggerEventBusMiddleware
SQLiteHistoryMirrorMiddleware

OtelTracingMiddleware from abxbus/OtelTracingMiddleware

​Interface

​Setup with EventBus

​Lifecycle behavior

​Custom middleware example

​Built-in implementations

Interface

Setup with EventBus

Lifecycle behavior

Custom middleware example

Built-in implementations