# BaseEvent
Source: https://abxbus.archivebox.io/api/baseevent
BaseEvent schema fields, lifecycle, and result helpers.
`BaseEvent` is the typed payload + runtime state object that flows through the bus.
Use subclassing (Python) or `BaseEvent.extend(...)` (TypeScript) to define event payload fields.
## Defining events
```python theme={null}
from abxbus import BaseEvent
class FooCreateEvent(BaseEvent[str]):
id: str | None = None
name: str
age: int
```
```ts theme={null}
import { BaseEvent } from 'abxbus'
import { z } from 'zod'
const FooCreateEvent = BaseEvent.extend('FooCreateEvent', {
id: z.string().nullable().optional(),
name: z.string(),
age: z.number(),
event_result_type: z.string(),
})
```
```go theme={null}
type FooCreateEvent struct {
ID *string `json:"id,omitempty"`
Name string `json:"name"`
Age int `json:"age"`
}
event, err := abxbus.Event(FooCreateEvent{
Name: "Ada",
Age: 37,
})
```
```rust theme={null}
use abxbus::event;
event! {
struct FooCreateEvent {
id: Option,
name: String,
age: i64,
event_result_type: String,
}
}
let event = bus.emit(FooCreateEvent {
id: None,
name: "Ada".to_string(),
age: 37,
..Default::default()
});
```
Queue semantics are split in each runtime:
* Immediate path: `await event.now()` (Python/TypeScript), `event.Now()` (Go), `event.now().await` (Rust)
* Queue-order path: `await event.wait()` (Python/TypeScript), `event.Wait()` (Go), `event.wait().await` (Rust)
`now()` and `wait()` accept the wait-shaping options (`first_result` and `timeout`). Result filtering and error policy live on `event_result()` / `event_results_list()`.
Python also supports `await event` as a Python-only shortcut for `await event.now()`. Other runtimes require the explicit `now()` / `wait()` methods.
In Go the same split is exposed as `event.Now()` for immediate queue-jump execution and `event.Wait()` for normal queue-order waiting.
In Rust the same split is exposed as `event.now().await` for immediate queue-jump execution and `event.wait().await` for normal queue-order waiting.
Unset event option fields are not hydrated onto the event at emit time; the bus that actually processes the event resolves its defaults at execution time.
## Core metadata fields
Common event metadata fields available in all runtimes:
* `event_id`: unique UUIDv7
* `event_type`: event name/type key
* `event_version`: payload version marker
* `event_result_type`: expected handler return schema/type
* `event_timeout`: per-event timeout override (`None`/`null` means use the current processing bus default at execution time; `0` disables)
* `event_handler_timeout`: per-handler timeout cap override
* `event_handler_slow_timeout`: per-handler slow warning threshold (`None`/`null` means use the current processing bus default; `0` disables)
* `event_slow_timeout`: per-event slow warning threshold (`None`/`null` means use the current processing bus default; `0` disables)
* `event_concurrency`: event scheduling mode override (`None`/`null`/zero-value mode means use the current processing bus default)
* `event_handler_concurrency`: handler scheduling mode override (`None`/`null`/zero-value mode means use the current processing bus default)
* `event_handler_completion`: handler completion strategy override (`None`/`null`/zero-value mode means use the current processing bus default)
## Runtime fields
* `event_status`: pending/started/completed
* `event_created_at`, `event_started_at`, `event_completed_at`
* `event_started_at` / `event_completed_at` are `None` (Python) / `null` (TypeScript) until set
* `event_parent_id` and `event_emitted_by_handler_id` are `None` / `null` when unset
* `event_blocks_parent_completion`: `True` / `true` only for linked children awaited from the handler that emitted them
* `event_path`: buses traversed
* `event_results`: per-handler result entries
* Child-event tracking (`event_children`/descendants)
## Completion model
Events are returned in pending state from `emit()`, then complete asynchronously.
```python theme={null}
pending = bus.emit(MyEvent())
completed = await pending.now()
completed_without_raise = await pending.now()
completed_in_queue_order = await pending.wait()
value = await completed.event_result()
```
```ts theme={null}
const pending = bus.emit(MyEvent({}))
const completed = await pending.now()
const completedWithoutRaise = await pending.now()
const completed_in_queue_order = await pending.wait()
const value = await completed.eventResult()
```
```go theme={null}
type MyEvent struct{}
pending := bus.Emit(MyEvent{})
completed, err := pending.Now()
if err != nil {
return err
}
completedInQueueOrder, err := pending.Wait()
if err != nil {
return err
}
value, err := completed.EventResult()
_ = completedInQueueOrder
```
```rust theme={null}
use futures::executor::block_on;
let pending = bus.emit(MyEvent {
..Default::default()
});
block_on(pending.now())?;
block_on(pending.wait())?;
let value = block_on(pending.event_result())?;
```
## Result access helpers
Result helpers share the same defaults in all runtimes:
* `raise_if_any=true`: raise when any handler result is an error.
* `raise_if_none=false`: return `None` / `undefined` / `nil` / `[]` when no result matches the filter.
* If every handler errors, only `raise_if_any=false` plus `raise_if_none=false` suppresses the error. Any other combination raises.
* A single handler error is raised as that error; multiple handler errors are raised as an aggregate/exception-group shape.
### First Result
```python theme={null}
completed = await event.now(first_result=True)
value = await completed.event_result()
```
```ts theme={null}
const value = await event.now({ first_result: true }).eventResult()
```
```go theme={null}
completed, err := event.Now(&abxbus.EventWaitOptions{FirstResult: true})
if err != nil {
return err
}
value, err := completed.EventResult()
```
```rust theme={null}
let completed = block_on(event.now_with_options(EventWaitOptions {
first_result: true,
..Default::default()
}))?;
let value = block_on(completed.event_result())?;
```
### All results
```python theme={null}
items = await event.event_results_list()
by_handler = {handler_id: result.result for handler_id, result in event.event_results.items()}
```
```ts theme={null}
const items = await event.eventResultsList()
const filtered = await event.eventResultsList({
include: (result) => typeof result === 'string',
raise_if_any: false,
raise_if_none: true,
})
const first = await event.eventResult()
const errors = event.event_errors
```
```go theme={null}
items, err := event.EventResultsList()
filtered, err := event.EventResultsList(
&abxbus.EventResultOptions{
Include: func(result any, eventResult *abxbus.EventResult) bool {
_, ok := result.(string)
return ok
},
RaiseIfAny: false,
RaiseIfNone: true,
},
)
```
```rust theme={null}
let items = block_on(event.event_results_list_with_options(EventResultOptions {
raise_if_any: false,
raise_if_none: false,
..EventResultOptions::default()
}))?;
let by_handler = event.event_results.read();
```
### Per-handler result entries
You can create/update a specific `EventResult` entry for a handler (useful for controlled seeding/rehydration flows).
```python theme={null}
pending = event.event_result_update(handler=handler_entry, eventbus=bus, status='pending')
pending.update(status='completed', result='seeded')
```
```ts theme={null}
const pending = event.eventResultUpdate(handler_entry, { eventbus: bus, status: 'pending' })
pending.update({ status: 'completed', result: 'seeded' })
```
```go theme={null}
pending := event.EventResultUpdate(handlerEntry, &abxbus.BaseEventResultUpdateOptions{
EventBus: bus,
EventResultUpdateOptions: abxbus.EventResultUpdateOptions{
Status: abxbus.EventResultPending,
},
})
pending.Update(&abxbus.EventResultUpdateOptions{
Status: abxbus.EventResultCompleted,
Result: "seeded",
})
```
```rust theme={null}
use abxbus::event_result::EventResultStatus;
use serde_json::json;
let pending = event.event_result_update(
&handler_entry,
Some(EventResultStatus::Pending),
None,
None,
None,
);
event.event_result_update(
&handler_entry,
Some(EventResultStatus::Completed),
Some(Some(json!("seeded"))),
None,
None,
);
```
## Resetting an event
You can create a fresh pending copy for re-emit. By default this regenerates ids/routing lineage, resets status plus processing timestamps (`event_started_at`, `event_completed_at`), clears handler results, and clears runtime attachment state. `event_created_at` remains the original creation timestamp. Pass a field group as `false` to preserve it on the returned copy.
```python theme={null}
fresh = event.event_reset()
preserved_results = event.event_reset(results=False)
```
```ts theme={null}
const fresh = event.eventReset()
const preservedResults = event.eventReset({ results: false })
```
```go theme={null}
fresh, err := event.EventReset()
disabled := false
preservedResults, err := event.EventReset(abxbus.EventResetOptions{Results: &disabled})
```
```rust theme={null}
let fresh = event.event_reset();
let preserved_results = event.event_reset_with_options(abxbus::base_event::EventResetOptions {
results: false,
..Default::default()
});
```
## Serialization
Events are JSON-serializable in all runtimes for bridge and cross-runtime workflows.
```python theme={null}
payload = event.model_dump(mode='json')
print(payload)
# {
# "event_id": "0190...",
# "event_type": "CreateUserEvent",
# "event_status": "pending",
# "event_result_type": {"type": "object", "...": "..."},
# "email": "someuser@example.com",
# "...": "..."
# }
restored = type(event).model_validate(payload)
```
```ts theme={null}
const payload = event.toJSON()
console.log(payload)
// {
// event_id: '0190...',
// event_type: 'CreateUserEvent',
// event_status: 'pending',
// event_result_type: { type: 'object', ... },
// email: 'someuser@example.com',
// ...
// }
const restored = BaseEvent.fromJSON(payload)
```
```go theme={null}
payload, err := event.ToJSON()
fmt.Println(string(payload))
// {
// "event_id": "0190...",
// "event_type": "CreateUserEvent",
// "event_status": "pending",
// "event_result_type": {"type": "object", "...": "..."},
// "email": "someuser@example.com",
// "...": "..."
// }
restored, err := abxbus.BaseEventFromJSON(payload)
```
```rust theme={null}
let payload = event.to_json_value();
println!("{}", serde_json::to_string_pretty(&payload)?);
// {
// "event_id": "0190...",
// "event_type": "CreateUserEvent",
// "event_status": "pending",
// "event_result_type": {"type": "object", "...": "..."},
// "email": "someuser@example.com",
// "...": "..."
// }
let restored: CreateUserEvent = serde_json::from_value(payload)?;
```
## Notes
* Reserved names are validated in all runtimes:
* runtime API names such as `bus`, `emit`, `now`, and `wait` cannot be provided as payload fields where they would collide with event methods/properties.
* Unknown `event_*` fields are rejected.
* Known built-in `event_*` fields (for example `event_timeout`) can still be intentionally overridden in event definitions.
* `model_*` is also reserved:
* Python: unknown `model_*` fields are rejected, but valid Pydantic namespace overrides (for example `model_config`) are allowed.
* TypeScript: any `model_*` field is rejected.
* `event_result_type` drives handler return validation in all runtimes.
* Parent-child tracking is automatic for `event.emit(...)` inside handlers. `bus.emit(...)` creates a top-level event with no parent link.
## Reserved Fields
```python theme={null}
from pydantic import ConfigDict
from abxbus import BaseEvent
class AllowedEvent(BaseEvent[None]):
event_timeout: float | None = 30 # allowed built-in event_* override
model_config = ConfigDict(extra='allow') # allowed Pydantic model_* override
# rejected: unknown reserved prefixes
class InvalidEvent(BaseEvent[None]):
event_some_field_we_dont_recognize: int = 1 # raises
model_something_random: int = 2 # raises
```
```ts theme={null}
import { BaseEvent } from 'abxbus'
import { z } from 'zod'
const AllowedEvent = BaseEvent.extend('AllowedEvent', {
event_timeout: 30, // allowed built-in event_* override
payload: z.string(),
})
// rejected: unknown event_* and all model_*
BaseEvent.extend('InvalidEvent', {
event_some_field_we_dont_recognize: 1, // throws
model_something_random: 2, // throws
})
```
```go theme={null}
timeout := 30.0
type AllowedEvent struct {
Payload string `json:"payload"`
EventTimeout *float64 `json:"event_timeout,omitempty"`
}
event, err := abxbus.Event(AllowedEvent{
Payload: "ok",
EventTimeout: &timeout, // allowed built-in event_* override
})
if err != nil {
return err
}
// Go payloads are passed as maps or typed structs, so unknown event_* / model_* keys
// should not be placed in the flattened payload. Use explicit BaseEvent fields instead.
```
```rust theme={null}
use abxbus::event;
event! {
struct AllowedEvent {
payload: String,
event_result_type: (),
}
}
let event = bus.emit(AllowedEvent {
payload: "ok".to_string(),
event_timeout: Some(30.0),
..Default::default()
});
// Rust validates unknown event_* / model_* keys during typed event construction
// and uses explicit BaseEventData fields for built-in runtime overrides.
```
# EventBus
Source: https://abxbus.archivebox.io/api/eventbus
EventBus constructors, configuration, and core methods.
`EventBus` is the central runtime for handler registration, event emit, history lookup, and lifecycle control.
## `EventBus(...)`
```python theme={null}
EventBus(
name: str | None = None,
event_concurrency: Literal['global-serial', 'bus-serial', 'parallel'] | str | None = None,
event_handler_concurrency: Literal['serial', 'parallel'] | str = 'serial',
event_handler_completion: Literal['all', 'first'] | str = 'all',
max_history_size: int | None = 100,
max_history_drop: bool = False,
event_timeout: float | None = 60.0,
event_slow_timeout: float | None = 300.0,
event_handler_slow_timeout: float | None = 30.0,
event_handler_detect_file_paths: bool = True,
middlewares: Sequence[EventBusMiddleware] | None = None,
)
```
```ts theme={null}
new EventBus(name?: string, options?: {
id?: string
max_history_size?: number | null
max_history_drop?: boolean
event_concurrency?: 'global-serial' | 'bus-serial' | 'parallel' | null
event_timeout?: number | null
event_slow_timeout?: number | null
event_handler_concurrency?: 'serial' | 'parallel' | null
event_handler_completion?: 'all' | 'first'
event_handler_slow_timeout?: number | null
event_handler_detect_file_paths?: boolean
middlewares?: Array EventBusMiddleware>
})
```
```go theme={null}
abxbus.NewEventBus(name string, options *abxbus.EventBusOptions)
type EventBusOptions struct {
ID string
MaxHistorySize *int
MaxHistoryDrop bool
EventConcurrency abxbus.EventConcurrencyMode
EventTimeout *float64
EventSlowTimeout *float64
EventHandlerConcurrency abxbus.EventHandlerConcurrencyMode
EventHandlerCompletion abxbus.EventHandlerCompletionMode
EventHandlerSlowTimeout *float64
EventHandlerDetectFilePaths *bool
Middlewares []abxbus.EventBusMiddleware
}
```
```rust theme={null}
use abxbus::{
event_bus::{EventBus, EventBusOptions},
types::{
EventConcurrencyMode,
EventHandlerCompletionMode,
EventHandlerConcurrencyMode,
},
};
let bus = EventBus::new_with_options(
Some("MyBus".to_string()),
EventBusOptions {
max_history_size: Some(100),
max_history_drop: false,
event_concurrency: EventConcurrencyMode::BusSerial,
event_timeout: Some(60.0),
event_slow_timeout: Some(300.0),
event_handler_concurrency: EventHandlerConcurrencyMode::Serial,
event_handler_completion: EventHandlerCompletionMode::All,
event_handler_slow_timeout: Some(30.0),
event_handler_detect_file_paths: true,
..EventBusOptions::default()
},
);
```
### Shared configuration semantics
| Option | Description |
| - | - |
| `name` | Human-readable bus name used in logs/labels. |
| `event_concurrency` | Event scheduling policy across queue processing (`global-serial`, `bus-serial`, `parallel`). |
| `event_handler_concurrency` | How handlers for one event execute (`serial` vs `parallel`). |
| `event_handler_completion` | Completion mode (`all` waits for all handlers, `first` resolves on first successful result). |
| `event_timeout` | Default outer timeout budget for event/handler execution. |
| `event_slow_timeout` | Slow-event warning threshold. |
| `event_handler_slow_timeout` | Slow-handler warning threshold. |
| `event_handler_detect_file_paths` | Whether to capture source path metadata for handlers. |
| `max_history_size` | Maximum retained history (`null` = unbounded, `0` = keep only in-flight). |
| `max_history_drop` | If `true`, drop oldest history entries when full; if `false`, reject new emits at limit. |
| `middlewares` | Ordered middleware instances (or middleware classes/constructors) that receive lifecycle hooks. |
Defaults are resolved at processing time on each bus, not copied onto the event at emit.
When event fields are unset (`None`/`null`/zero-value mode), the current processing bus applies its own defaults, so forwarded events can inherit the target bus defaults.
## Runtime state
All implementations expose equivalent runtime state:
* Bus identity: `id`, `name`, `label`
* Registered handlers and indexes
* Event history and pending queue
* In-flight tracking
* Locking/concurrency runtime objects
## `on(...)`
Registers a handler for an event key (`EventClass`, event type string, or `'*'`).
```python theme={null}
bus.on(UserEvent, handler)
bus.on('UserEvent', handler)
bus.on('*', wildcard_handler)
```
```ts theme={null}
bus.on(UserEvent, handler)
bus.on('UserEvent', handler)
bus.on('*', wildcardHandler)
```
```go theme={null}
bus.On("UserEvent", "handler", handler, nil)
bus.OnEventName("*", "wildcard_handler", wildcardHandler, nil)
timeout := 5.0
bus.On("UserEvent", "custom", handler, &abxbus.EventHandler{
HandlerTimeout: &timeout,
})
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus, event_handler::EventHandlerOptions};
use serde_json::json;
event! {
struct UserEvent {
user_id: String,
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("ApiBus".to_string()));
let entry = bus.on(UserEvent, |event: UserEvent| async move {
Ok(json!(event.user_id))
});
bus.on_with_options(
UserEvent,
"custom",
EventHandlerOptions {
handler_timeout: Some(5.0),
..EventHandlerOptions::default()
},
|_event: UserEvent| async move { Ok(serde_json::Value::Null) },
);
```
## `off(...)`
Unregisters handlers by event key, handler function/reference, or handler id.
```python theme={null}
bus.off(UserEvent, handler)
bus.off(UserEvent) # remove all handlers for UserEvent
bus.off('*') # remove all wildcard handlers
```
```ts theme={null}
bus.off(UserEvent, handler)
bus.off(UserEvent)
bus.off('*')
```
```go theme={null}
handler := bus.On("UserEvent", "handler", onUserEvent, nil)
bus.Off("UserEvent", handler)
bus.Off("UserEvent", handler.ID)
bus.Off("*", nil)
```
```rust theme={null}
use abxbus::typed::EventSpec;
let handler = bus.on(UserEvent, |_event: UserEvent| async move {
Ok(serde_json::Value::Null)
});
bus.off(UserEvent::event_type, Some(&handler.id));
bus.off(UserEvent::event_type, None);
```
## `emit(...)`
`emit(...)` enqueues synchronously and returns the pending event immediately.
```python theme={null}
event = bus.emit(MyEvent(data='x'))
result = await event.event_result()
```
```ts theme={null}
const event = bus.emit(MyEvent({ data: 'x' }))
const result = await event.now({ first_result: true }).eventResult()
```
```go theme={null}
type MyEvent struct {
Data string `json:"data"`
}
event := bus.Emit(MyEvent{Data: "x"})
result, err := event.EventResult()
if err != nil {
return err
}
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
event! {
struct MyEvent {
data: String,
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("EmitBus".to_string()));
let event = bus.emit(MyEvent { data: "x".to_string(), ..Default::default() });
let result = block_on(event.event_result())?;
```
## `find(...)`
`find(...)` supports history lookup, optional future waiting, predicate filtering, and parent/child scoping.
```python theme={null}
event = await bus.find(ResponseEvent) # history lookup by default
future = await bus.find(ResponseEvent, past=False, future=5)
child = await bus.find(ChildEvent, child_of=parent_event, future=5)
```
```ts theme={null}
const event = await bus.find(ResponseEvent)
const future = await bus.find(ResponseEvent, { past: false, future: 5 })
const child = await bus.find(ChildEvent, { child_of: parentEvent, future: 5 })
```
```go theme={null}
event, err := bus.FindEventName("ResponseEvent", nil, nil)
future, err := bus.FindEventName("ResponseEvent", nil, &abxbus.FindOptions{
Past: false,
Future: 5.0,
})
child, err := bus.FindEventName("ChildEvent", nil, &abxbus.FindOptions{
ChildOf: parentEvent,
Future: 5.0,
})
```
```rust theme={null}
use abxbus::event_bus::FindOptions;
use futures::executor::block_on;
let event = block_on(bus.find("ResponseEvent", true, None, None));
let future = block_on(bus.find("ResponseEvent", false, Some(5.0), None));
let child = block_on(bus.find_with_options(
"ChildEvent",
FindOptions {
child_of: Some(parent_event.clone()),
future: Some(5.0),
..FindOptions::default()
},
));
```
## `filter(...)`
`filter(...)` takes the same arguments as `find(...)` but returns the list of all matching
events (newest to oldest) instead of just the first match. Accepts an additional `limit`
argument to cap the result count.
```python theme={null}
recent = await bus.filter(ResponseEvent, past=10, future=False, limit=5)
```
```ts theme={null}
const recent = await bus.filter(ResponseEvent, { past: 10, future: false, limit: 5 })
```
```go theme={null}
limit := 5
recent, err := bus.FilterEventName("ResponseEvent", nil, &abxbus.FilterOptions{
Past: 10.0,
Future: false,
Limit: &limit,
})
recentWithPredicate, err := bus.FilterEventName(
"ResponseEvent",
func(event *abxbus.BaseEvent) bool {
return event.Payload["ok"] == true
},
&abxbus.FilterOptions{Past: 10.0, Future: false, Limit: &limit},
)
```
```rust theme={null}
use abxbus::event_bus::FilterOptions;
use futures::executor::block_on;
use serde_json::json;
use std::collections::HashMap;
let recent = block_on(bus.filter("ResponseEvent", true, None, None, Some(5)));
let filtered = block_on(bus.filter_with_options(
"ResponseEvent",
FilterOptions {
past_window: Some(10.0),
where_filter: Some(HashMap::from([("ok".to_string(), json!(true))])),
limit: Some(5),
..FilterOptions::default()
},
));
```
## Lifecycle helpers
### Wait for idle
```python theme={null}
await bus.wait_until_idle()
await bus.wait_until_idle(timeout=5)
```
```ts theme={null}
await bus.waitUntilIdle()
await bus.waitUntilIdle(5)
```
```go theme={null}
bus.WaitUntilIdle(nil)
timeout := 5.0
bus.WaitUntilIdle(&timeout)
```
```rust theme={null}
use futures::executor::block_on;
block_on(bus.wait_until_idle(None));
block_on(bus.wait_until_idle(Some(5.0)));
```
### Parent/child relationship checks
```python theme={null}
bus.event_is_child_of(child_event, parent_event)
bus.event_is_parent_of(parent_event, child_event)
```
```ts theme={null}
bus.eventIsChildOf(childEvent, parentEvent)
bus.eventIsParentOf(parentEvent, childEvent)
```
```go theme={null}
bus.EventIsChildOf(childEvent, parentEvent)
bus.EventIsParentOf(parentEvent, childEvent)
```
```rust theme={null}
bus.event_is_child_of(&child_event, &parent_event);
bus.event_is_parent_of(&parent_event, &child_event);
```
### Execution pipeline
All runtimes expose the same public processing contract:
* `emit(...)` accepts the event, records it in bus history, and queues it for processing.
* `find(...)` can observe accepted events before handlers finish.
* Event-level timeout and slow-warning settings apply to the whole event.
* Handler-level timeout and slow-warning settings apply to each handler.
* `event_concurrency`, `event_handler_concurrency`, and `event_handler_completion` choose queueing and completion behavior.
* Handler return values and errors are stored in `event_results`; use the result helpers to retrieve typed raw values.
### Serialization and teardown
All runtimes can serialize the entire bus state (config, handlers metadata, event history, pending queue), restore it, re-attach handler callables, and continue processing.
```python theme={null}
from abxbus import BaseEvent, EventBus
class ResumeTickEvent(BaseEvent[None]):
pass
# 1) Serialize full bus state
print(bus.model_dump_json(indent=2))
# {
# "id": "018f...",
# "name": "SerializableBus",
# "handlers": {
# "018f...": {"id": "018f...", "handler_name": "app.handlers.on_user_created", "event_pattern": "UserCreatedEvent"}
# },
# "event_history": {
# "0190...": {"event_id": "0190...", "event_type": "UserCreatedEvent", "...": "..."}
# },
# "pending_event_queue": ["0190..."]
# }
# 2) Rehydrate state
bus = EventBus.validate(bus.model_dump_json())
# 3) Re-link runtime callables by handler id
bus.handlers.get("018f...").handler = on_user_created
bus.handlers.get("018g...").handler = on_user_deleted
bus.handlers.get("018h...").handler = on_user_updated
# 4) Resume processing (emit starts the runloop and drains restored pending events)
bus.emit(ResumeTickEvent())
await bus.wait_until_idle()
# 5) Teardown when done
await bus.destroy() # destroy immediately and clear by default
await bus.destroy(clear=False) # preserve handlers/history if you need to inspect
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
const ResumeTickEvent = BaseEvent.extend('ResumeTickEvent', {})
// 1) Serialize full bus state
console.log(bus.toJSON())
// {
// id: '018f...',
// name: 'SerializableBus',
// handlers: {
// '018f...': { id: '018f...', handler_name: 'onUserCreated', event_pattern: 'UserCreatedEvent' }
// },
// event_history: {
// '0190...': { event_id: '0190...', event_type: 'UserCreatedEvent', ... }
// },
// pending_event_queue: ['0190...']
// }
// 2) Rehydrate state
bus = EventBus.fromJSON(bus.toJSON())
// 3) Re-link runtime callables by handler id
bus.handlers.get('018f...')!.handler = onUserCreated
bus.handlers.get('018g...')!.handler = onUserDeleted
bus.handlers.get('018h...')!.handler = onUserUpdated
// 4) Resume processing (emit starts the runloop and drains restored pending events)
bus.emit(ResumeTickEvent({}))
await bus.waitUntilIdle()
// 5) Teardown when done
await bus.destroy({ clear: true })
```
```go theme={null}
type ResumeTickEvent struct{}
resumeTick, err := abxbus.Event(ResumeTickEvent{})
if err != nil {
panic(err)
}
// 1) Serialize full bus state
payload, err := bus.ToJSON()
fmt.Println(string(payload))
// {
// "id": "018f...",
// "name": "SerializableBus",
// "handlers": {
// "018f...": {"id": "018f...", "handler_name": "onUserCreated", "event_pattern": "UserCreatedEvent"}
// },
// "event_history": {
// "0190...": {"event_id": "0190...", "event_type": "UserCreatedEvent", "...": "..."}
// },
// "pending_event_queue": ["0190..."]
// }
// 2) Rehydrate state
bus, err = abxbus.EventBusFromJSON(payload)
// 3) Re-link runtime callables by handler id
bus.On("UserCreatedEvent", "onUserCreated", onUserCreated, bus.Handlers["018f..."])
bus.On("UserDeletedEvent", "onUserDeleted", onUserDeleted, bus.Handlers["018g..."])
bus.On("UserUpdatedEvent", "onUserUpdated", onUserUpdated, bus.Handlers["018h..."])
// 4) Resume processing
bus.Emit(resumeTick)
bus.WaitUntilIdle(nil)
// 5) Teardown when done
bus.Destroy()
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
use serde_json::json;
event! { struct ResumeTickEvent { event_result_type: serde_json::Value } }
event! { struct UserCreatedEvent { event_result_type: serde_json::Value } }
event! { struct UserDeletedEvent { event_result_type: serde_json::Value } }
event! { struct UserUpdatedEvent { event_result_type: serde_json::Value } }
let resume_tick = ResumeTickEvent { ..Default::default() };
// 1) Serialize full bus state
let payload = bus.to_json_value();
println!("{}", serde_json::to_string_pretty(&payload).unwrap());
// {
// "id": "018f...",
// "name": "SerializableBus",
// "handlers": {
// "018f...": {"id": "018f...", "handler_name": "on_user_created", "event_pattern": "UserCreatedEvent"}
// },
// "event_history": {
// "0190...": {"event_id": "0190...", "event_type": "UserCreatedEvent", "...": "..."}
// },
// "pending_event_queue": ["0190..."]
// }
// 2) Rehydrate state
let bus = EventBus::from_json_value(payload);
// 3) Re-link runtime callables by handler id/name
bus.on(UserCreatedEvent, |_event: UserCreatedEvent| async move { Ok(json!("created")) });
bus.on(UserDeletedEvent, |_event: UserDeletedEvent| async move { Ok(json!("deleted")) });
bus.on(UserUpdatedEvent, |_event: UserUpdatedEvent| async move { Ok(json!("updated")) });
// 4) Resume processing
bus.emit(resume_tick);
assert!(block_on(bus.wait_until_idle(Some(1.0))));
// 5) Teardown when done
bus.destroy();
```
## `destroy(...)`
Destroy stops runtime work and optionally clears bus-owned state.
```python theme={null}
await bus.destroy()
await bus.destroy(clear=False)
```
```ts theme={null}
await bus.destroy()
await bus.destroy({ clear: false })
```
```go theme={null}
bus.Destroy()
bus.DestroyWithOptions(&abxbus.EventBusDestroyOptions{Clear: false})
```
```rust theme={null}
use abxbus::event_bus::DestroyOptions;
bus.destroy();
bus.destroy_with_options(DestroyOptions { clear: false, ..Default::default() });
```
* `clear` defaults to `true` in every runtime.
* With `clear=true`, destroy is terminal and clears handlers, history, queues, in-flight tracking, waiters, locks, and dispatch/context state so the bus can be released.
* With `clear=false`, destroy is still terminal and the bus cannot be used again; it only preserves handlers/history for inspection.
* Destroying one bus only clears that bus's local state; shared event objects, handler functions, and other buses are not destroyed.
## Timeout and precedence
Shared precedence model:
1. Handler override
2. Event override
3. Bus default
Effective handler timeout is capped by event timeout when both are set.
# EventBusMiddleware
Source: https://abxbus.archivebox.io/api/eventbusmiddleware
Base middleware interface for EventBus lifecycle hooks.
`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 theme={null}
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:
...
```
```ts theme={null}
import type { BaseEvent, EventBus, EventHandler, EventResult, EventStatus } from 'abxbus'
export interface EventBusMiddleware {
onEventChange?(eventbus: EventBus, event: BaseEvent, status: EventStatus): void | Promise
onEventResultChange?(
eventbus: EventBus,
event: BaseEvent,
event_result: EventResult,
status: EventStatus
): void | Promise
onBusHandlersChange?(eventbus: EventBus, handler: EventHandler, registered: boolean): void | Promise
}
```
```go theme={null}
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
}
```
```rust theme={null}
// 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 theme={null}
from abxbus import EventBus
from abxbus.middlewares import EventBusMiddleware
class AnalyticsMiddleware(EventBusMiddleware):
pass
bus = EventBus('AppBus', middlewares=[AnalyticsMiddleware])
```
```ts theme={null}
import { EventBus } from 'abxbus'
import type { EventBusMiddleware } from 'abxbus/middlewares'
class AnalyticsMiddleware implements EventBusMiddleware {}
const bus = new EventBus('AppBus', { middlewares: [AnalyticsMiddleware] })
```
```go theme={null}
type AnalyticsMiddleware struct {
abxbus.EventBusMiddlewareBase
}
bus := abxbus.NewEventBus("AppBus", &abxbus.EventBusOptions{
Middlewares: []abxbus.EventBusMiddleware{&AnalyticsMiddleware{}},
})
```
```rust theme={null}
// 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 theme={null}
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])
```
```ts theme={null}
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 {
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 {
const action = registered ? 'registered' : 'unregistered'
console.log(eventbus.label, handler.id, action)
}
}
const bus = new EventBus('AppBus', { middlewares: [AnalyticsMiddleware] })
```
```go theme={null}
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{}},
})
```
```rust theme={null}
// Not implemented in the Rust crate yet.
```
## Built-in implementations
* `OtelTracingMiddleware`
* `AutoErrorEventMiddleware`
* `AutoReturnEventMiddleware`
* `AutoHandlerChangeEventMiddleware`
* `WALEventBusMiddleware`
* `LoggerEventBusMiddleware`
* `SQLiteHistoryMirrorMiddleware`
* `OtelTracingMiddleware` from `abxbus/OtelTracingMiddleware`
* Middleware interface only.
Go intentionally keeps middleware implementations that require optional third-party packages out of the core module so those packages do not become hard dependencies.
* Rust middleware implementations are not exposed yet.
# EventHandler
Source: https://abxbus.archivebox.io/api/eventhandler
EventHandler metadata and registration records.
`EventHandler` is the serializable metadata record for a registered handler function.
You receive handler entries from `bus.on(...)`, can remove them with `bus.off(...)`, and see them in handler-change middleware/hooks.
## Common fields
* `id`: stable handler id
* `handler_name`: callable/function name
* `handler_file_path`: source path metadata (`str | None` in Python, `string | null` in TypeScript)
* `handler_timeout`: optional per-handler timeout override
* `handler_slow_timeout`: optional slow-warning override
* `handler_registered_at`
* `event_pattern`: subscribed key (`EventType` or `'*'`)
* `eventbus_name`, `eventbus_id`
## `bus.on(...)` and `bus.off(...)`
```python theme={null}
entry = bus.on(MyEvent, handler)
bus.off(MyEvent, entry)
```
```ts theme={null}
const entry = bus.on(MyEvent, handler)
bus.off(MyEvent, entry)
```
```go theme={null}
type MyEvent struct{}
handler := func(event MyEvent) {}
entry := bus.On(handler)
bus.Off("MyEvent", entry)
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use abxbus::typed::EventSpec;
event! {
struct MyEvent {
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("HandlerBus".to_string()));
let entry = bus.on(MyEvent, |_event: MyEvent| async move {
Ok(serde_json::Value::Null)
});
bus.off(MyEvent::event_type, Some(&entry.id));
```
## Serialization
```python theme={null}
payload = entry.model_dump(mode='json', exclude={'handler'})
print(payload)
# {
# "id": "018f...",
# "handler_name": "app.handlers.on_user_created",
# "event_pattern": "UserCreatedEvent",
# "eventbus_name": "AuthBus",
# "eventbus_id": "018f...",
# "...": "..."
# }
restored = EventHandler.from_json_dict(payload, handler=real_handler)
```
```ts theme={null}
const payload = entry.toJSON()
console.log(payload)
// {
// id: '018f...',
// handler_name: 'onUserCreated',
// event_pattern: 'UserCreatedEvent',
// eventbus_name: 'AuthBus',
// eventbus_id: '018f...',
// ...
// }
const restored = EventHandler.fromJSON(payload, realHandler)
```
```go theme={null}
payload, err := entry.ToJSON()
fmt.Println(string(payload))
// {
// "id": "018f...",
// "handler_name": "on_user_created",
// "event_pattern": "UserCreatedEvent",
// "eventbus_name": "AuthBus",
// "eventbus_id": "018f...",
// ...
// }
restored, err := abxbus.EventHandlerFromJSON(payload, realHandler)
```
```rust theme={null}
let payload = entry.to_json_value();
println!("{}", serde_json::to_string_pretty(&payload)?);
// {
// "id": "018f...",
// "handler_name": "on_user_created",
// "event_pattern": "UserCreatedEvent",
// "eventbus_name": "AuthBus",
// "eventbus_id": "018f...",
// ...
// }
let restored = abxbus::event_handler::EventHandler::from_json_value(payload);
```
## Notes
* Function bodies are not serialized.
* Rehydration restores metadata; execution behavior requires re-binding a real callable.
# EventHistory
Source: https://abxbus.archivebox.io/api/eventhistory
EventHistory ordered store API for lookup, retention, and trimming.
`EventHistory` is the ordered event store owned by each `EventBus` as `bus.event_history`.
Rust stores the same history inside `EventBus` runtime state and exposes it through `EventBus` history methods rather than a standalone public `EventHistory` type.
It is the canonical history backend in each runtime and exposes ordered access, lookup (`find`), and retention controls.
## Common fields
* `max_history_size`: max retained events (`None`/`null` means unbounded, `0` keeps only in-flight visibility)
* `max_history_drop`: whether to trim oldest entries when over limit
* Ordered mapping keyed by `event_id`
## Mapping interface
`EventHistory` behaves like an ordered mapping in Python/TypeScript; Go exposes equivalent explicit methods.
```python theme={null}
from abxbus import EventHistory
history = EventHistory(max_history_size=100, max_history_drop=True)
history[event.event_id] = event
exists = event.event_id in history
loaded = history.get(event.event_id)
count = len(history)
for event_id, item in history.items():
print(event_id, item.event_type)
```
```ts theme={null}
import { EventHistory } from 'abxbus'
const history = new EventHistory({ max_history_size: 100, max_history_drop: true })
history.set(event.event_id, event)
const exists = history.has(event.event_id)
const loaded = history.get(event.event_id)
const count = history.size
for (const [event_id, item] of history) {
console.log(event_id, item.event_type)
}
```
```go theme={null}
maxSize := 100
history := abxbus.NewEventHistory(&maxSize, true)
history.AddEvent(event)
exists := history.Has(event.EventID)
loaded := history.GetEvent(event.EventID)
count := history.Size()
for _, item := range history.Values() {
fmt.Println(item.EventID, item.EventType)
}
```
```rust theme={null}
let bus = EventBus::new_with_history(Some("HistoryBus".to_string()), Some(100), true);
let event = bus.emit(SomeEvent {
..Default::default()
});
block_on(event.now());
let count = bus.event_history_size();
let ids = bus.event_history_ids();
let loaded = bus.runtime_payload_for_test().get(&ids[0]).cloned();
```
## Accessor methods
All runtimes expose explicit accessors in addition to mapping methods where the language supports them.
* `add_event(event)`
* `get_event(event_id)`
* `remove_event(event_id)`
* `has_event(event_id)`
* `find(...)`
* `filter(...)`
* `trim_event_history(...)`
* `addEvent(event)`
* `getEvent(event_id)`
* `removeEvent(event_id)`
* `hasEvent(event_id)`
* `find(...)`
* `filter(...)`
* `trimEventHistory(...)`
* `AddEvent(event)`
* `GetEvent(event_id)`
* `RemoveEvent(event_id)`
* `Has(event_id)`
* `Find(...)`
* `Filter(...)`
* `TrimEventHistory(...)`
* `event_history_size()`
* `event_history_ids()`
* `find(...)`
* `filter(...)`
* `max_history_size()`
* `max_history_drop()`
## `find(...)`
`EventHistory.find(...)` supports:
* event pattern (`'EventType'`, class, or `'*'`)
* `where` predicate filtering
* field equality filters (`**event_fields` in Python, options object fields in TypeScript)
* `past` / `future` windows
* `child_of` constraints
`EventBus.find(...)` delegates to `event_history.find(...)`. Future waiter ownership remains on `EventBus`; the bus injects the callback used for future waits.
## `filter(...)`
Same as `find(...)` but returns the list of all matching events (newest to oldest)
instead of just the first match. Accepts an additional `limit` argument to cap the
result count. `find(...)` is implemented as `filter(..., limit=1)` returning the first
result or `None` / `null`.
## Retention and trim methods
* `trim_event_history` / `trimEventHistory` applies configured retention policy.
* `on_remove` callbacks let the caller run cleanup side effects when history entries are removed.
## Constructor options
```python theme={null}
history = EventHistory(max_history_size=100, max_history_drop=False)
```
```ts theme={null}
const history = new EventHistory({ max_history_size: 100, max_history_drop: false })
```
```go theme={null}
maxSize := 100
history := abxbus.NewEventHistory(&maxSize, false)
```
```rust theme={null}
let bus = EventBus::new_with_history(Some("HistoryBus".to_string()), Some(100), false);
```
# EventResult
Source: https://abxbus.archivebox.io/api/eventresult
EventResult fields, status, and handler execution results.
Each handler execution for an event produces one `EventResult`.
You usually access results through `event.event_results` (or high-level event helper methods), but this page documents the underlying object.
## Common fields
* `id`: unique result id
* `status`: `pending | started | completed | error`
* `result`: handler return value (typed by event result schema/type)
* `error`: captured exception/error when handler fails
* `started_at`, `completed_at` (`None`/`null` until the handler starts/completes)
* `event_children`: child events emitted from inside this handler execution
* Handler metadata (`handler_id`, `handler_name`, `handler_file_path`, bus label/id/name)
## Await semantics
Awaiting an `EventResult` resolves to handler return value or raises captured failure in Python.
TypeScript and Go expose the stored result/error fields directly; Go's `Wait(...)` only waits for settlement.
```python theme={null}
entry = event.event_results[some_handler_id]
value = await entry
```
```ts theme={null}
const [, entry] = Array.from(event.event_results.entries())[0]
const value = entry.result
```
```go theme={null}
for _, entry := range event.EventResults {
if err := entry.Wait(); err != nil {
return err
}
value := entry.Result
_ = value
break
}
```
```rust theme={null}
let entry = event
.event_results
.read()
.values()
.next()
.cloned()
.expect("handler result");
let value = entry.result.clone();
```
## Lifecycle
Every handler entry follows the same public lifecycle in each runtime:
1. `pending`: the event has accepted the handler result slot, but the handler has not started.
2. `started`: the handler callable is running.
3. `completed`: the handler returned a valid value or `None` / `null` / `nil`.
4. `error`: the handler raised, returned an invalid value, timed out, or was cancelled/aborted by completion policy.
Use `event.event_results` for the per-handler mapping and `event_result()` / `eventResult()` / `EventResult()` / `event_result().await` when you only need the first typed raw result value.
## Serialization
```python theme={null}
payload = entry.model_dump(mode='json')
print(payload)
# {
# "id": "0190...",
# "status": "completed",
# "event_id": "018f...",
# "handler_id": "018g...",
# "handler_name": "on_user_created",
# "result": {"user_id": "u_123"},
# "error": None,
# "...": "..."
# }
restored = EventResult.model_validate(payload)
```
```ts theme={null}
const payload = entry.toJSON()
console.log(payload)
// {
// id: '0190...',
// status: 'completed',
// event_id: '018f...',
// handler: { id: '018g...', handler_name: 'onUserCreated', ... },
// result: { user_id: 'u_123' },
// error: undefined,
// ...
// }
const restored = EventResult.fromJSON(event, payload)
```
```go theme={null}
payload, err := json.Marshal(entry)
fmt.Println(string(payload))
// {
// "id": "0190...",
// "status": "completed",
// "event_id": "018f...",
// "handler_id": "018g...",
// "handler_name": "on_user_created",
// "result": {"user_id": "u_123"},
// "error": null,
// ...
// }
restored, err := abxbus.EventResultFromJSON(payload)
```
```rust theme={null}
let payload = serde_json::to_value(&entry)?;
println!("{}", serde_json::to_string_pretty(&payload)?);
// {
// "id": "0190...",
// "status": "completed",
// "event_id": "018f...",
// "handler_id": "018g...",
// "handler_name": "on_user_created",
// "result": {"user_id": "u_123"},
// "error": null,
// ...
// }
let restored: abxbus::event_result::EventResult = serde_json::from_value(payload)?;
```
# retry
Source: https://abxbus.archivebox.io/api/retry
Retry decorator/higher-order wrapper for async and sync functions and handlers.
`retry` adds per-attempt timeout, retry/backoff, and optional semaphore-based concurrency control around async and sync callables. It works for ordinary functions without creating an `EventBus`.
Sync callables stay sync and block synchronously for retry sleeps and semaphore waits. Async callables stay async and await retry sleeps and semaphore waits.
Supported callable forms:
* Python: sync and async functions, methods, and lambdas via `@retry(...)` or `retry(...)(fn)`.
* TypeScript: sync and async functions and arrow functions via `retry(...)(fn)`, plus sync and async class methods via `@retry(...)`.
* Rust: sync and async free functions and methods via `abxbus::retry!`; call those retried functions from Rust closures such as EventBus handlers.
* Go: retry is not implemented yet.
## Signature
```python theme={null}
def retry(
retry_after: float = 0,
max_attempts: int = 1,
timeout: float | None = None,
slow_timeout: float | None = None,
retry_on_errors: list[type[Exception] | re.Pattern[str]] | tuple[type[Exception] | re.Pattern[str], ...] | None = None,
retry_backoff_factor: float = 1.0,
semaphore_limit: int | None = None,
semaphore_name: str | Callable[..., str] | None = None,
semaphore_lax: bool = True,
semaphore_scope: Literal['multiprocess', 'global', 'class', 'instance'] = 'global',
semaphore_timeout: float | None = None,
) -> Callable[[Callable[P, T]], Callable[P, T]]
```
```ts theme={null}
retry({
max_attempts?: number, // default: 1
retry_after?: number, // default: 0 (seconds)
retry_backoff_factor?: number, // default: 1.0
retry_on_errors?: Array<(new (...args) => Error) | RegExp | string>, // default: retry any error
timeout?: number | null, // default: no per-attempt timeout
slow_timeout?: number | null, // default: disabled
semaphore_limit?: number | null, // default: no semaphore limit
semaphore_name?: string | ((...args: any[]) => string) | null, // default: function name
semaphore_lax?: boolean, // default: true
semaphore_scope?: 'multiprocess' | 'global' | 'class' | 'instance', // default: 'global'
semaphore_timeout?: number | null, // default: derived when timeout + limit are set
})
```
```rust theme={null}
abxbus::retry! {
max_attempts = 3,
retry_after = 1.0,
retry_backoff_factor = 2.0,
timeout = 5.0,
slow_timeout = 20.0,
retry_if = should_retry,
semaphore_limit = 2,
semaphore_name = "api",
semaphore_lax = true,
semaphore_scope = "global",
semaphore_timeout = 10.0;
async fn run(&self) -> Result {
/* body */
}
}
```
```go theme={null}
// Not implemented in abxbus-go yet.
//
// Use bus/event/handler timeout fields for outer execution budgets today.
// Per-attempt retry/backoff/semaphore behavior should be wrapped inside
// your Go handler until retry parity is implemented.
```
## Options
| Option | Description |
| - | - |
| `max_attempts` | Total attempts including the first call (`1` disables retries). |
| `retry_after` | Base delay between retries, in seconds. |
| `retry_backoff_factor` | Delay multiplier applied after each failed attempt. |
| `retry_on_errors` | Optional matcher list to restrict which errors are retried. |
| `timeout` | Per-attempt timeout in seconds (`None`/`undefined` means no per-attempt timeout). |
| `slow_timeout` | Warning threshold for a decorated call, in seconds. Warnings are throttled to at most one every 2 seconds per decorated method/function. |
| `semaphore_limit` | Max concurrent executions sharing the same semaphore. |
| `semaphore_name` | Semaphore key (string or function deriving a key from call args). |
| `semaphore_scope` | Semaphore sharing scope (`multiprocess`, `global`, `class`, `instance`). |
| `semaphore_timeout` | Max wait time for semaphore acquisition before timeout/lax fallback. |
| `semaphore_lax` | If true, continue execution without semaphore limit when acquisition times out. |
## Example: Standalone function
```python theme={null}
from abxbus.retry import retry
@retry(max_attempts=3, retry_after=1, timeout=5)
async def fetch_with_retry(url: str) -> dict:
return await fetch_json(url)
```
```ts theme={null}
import { retry } from 'abxbus/retry'
async function fetchJsonWithRetry(url: string): Promise> {
return await fetchJson(url)
}
const fetchWithRetry = retry({ max_attempts: 3, retry_after: 1, timeout: 5 })(fetchJsonWithRetry)
```
```rust theme={null}
#[derive(Debug)]
enum ApiError {
Retry(abxbus::retry::RetryError),
Transport(String),
}
impl From for ApiError {
fn from(error: abxbus::retry::RetryError) -> Self {
Self::Retry(error)
}
}
abxbus::retry! {
max_attempts = 3, retry_after = 1.0, timeout = 5.0;
async fn fetch_with_retry(url: String) -> Result {
fetch_json(url).await.map_err(ApiError::Transport)
}
}
```
## Example: Inline wrapper
```python theme={null}
from abxbus import EventBus, BaseEvent
from abxbus.retry import retry
class FetchEvent(BaseEvent[dict]):
url: str
bus = EventBus('AppBus')
async def fetch_with_retry(event: FetchEvent) -> dict:
return await fetch_json(event.url)
bus.on(
FetchEvent,
retry(max_attempts=3, retry_after=1, timeout=5)(fetch_with_retry),
)
```
```ts theme={null}
import { BaseEvent, EventBus, retry } from 'abxbus'
import { z } from 'zod'
const FetchEvent = BaseEvent.extend('FetchEvent', {
url: z.string(),
event_result_type: z.record(z.string(), z.unknown()),
})
const bus = new EventBus('AppBus')
bus.on(
FetchEvent,
retry({ max_attempts: 3, retry_after: 1, timeout: 5 })(async (event) => {
return await fetchJson(event.url)
})
)
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
struct Api;
impl Api {
abxbus::retry! {
max_attempts = 3, retry_after = 1.0, timeout = 5.0;
async fn fetch_with_retry(&self, event: FetchEvent) -> Result {
fetch_json(event.url).await.map_err(ApiError::Transport)
}
}
}
let api = Api;
let bus = EventBus::new(Some("AppBus".to_string()));
bus.on(FetchEvent, move |event: FetchEvent| async move {
api.fetch_with_retry(event).await
});
```
```go theme={null}
// retry decorators/method wrappers are not implemented in abxbus-go yet.
//
// Put retry/backoff code in a helper function or inside the method body,
// then register that method as a normal EventBus handler.
```
## Example: Decorated class method
```python theme={null}
from abxbus.retry import retry
class ApiService:
@retry(max_attempts=4, retry_after=1, timeout=10, semaphore_limit=2, semaphore_scope='class')
async def get_user(self, user_id: str) -> dict:
return await call_remote_api(user_id)
```
```ts theme={null}
import { retry } from 'abxbus/retry'
class ApiService {
@retry({ max_attempts: 4, retry_after: 1, timeout: 10, semaphore_limit: 2, semaphore_scope: 'class' })
async getUser(userId: string): Promise> {
return await callRemoteApi(userId)
}
}
```
Without decorators, use the same closure-based wrapper on a normal function:
```ts theme={null}
import { retry } from 'abxbus/retry'
async function getUser(userId: string): Promise> {
return await callRemoteApi(userId)
}
const getUserWithRetry = retry({ max_attempts: 4, retry_after: 1, timeout: 10, semaphore_limit: 2 })(getUser)
```
The same `@retry(...)` method syntax works with TypeScript's standard decorators and with legacy `experimentalDecorators` method decorators. For legacy decorators, enable `"experimentalDecorators": true` in `tsconfig.json`; the method keeps its original sync or async return type.
```rust theme={null}
struct ApiService;
impl ApiService {
abxbus::retry! {
max_attempts = 4,
retry_after = 1.0,
timeout = 10.0,
semaphore_limit = 2,
semaphore_scope = "class";
async fn get_user(&self, user_id: String) -> Result {
call_remote_api(user_id).await.map_err(ApiError::Transport)
}
}
}
```
```go theme={null}
// retry decorators/method wrappers are not implemented in abxbus-go yet.
//
// Put retry/backoff code in a helper function or inside the method body,
// then register that method as a normal EventBus handler.
```
## Behavior
* Semaphore acquisition happens once per call, then all retry attempts run within that acquired slot.
* Backoff delay per retry is: `retry_after * retry_backoff_factor^(attempt - 1)`.
* Retries stop immediately when the thrown error does not match `retry_on_errors`.
* Bus/event timeouts act as outer execution budgets; `retry.timeout` is per-attempt.
* Sync wrappers return synchronously and block synchronously for retry sleeps and semaphore waits.
* Rust retry functions return `Result` where `E: From`.
## Runtime differences
* Python and TypeScript both support `multiprocess`, `global`, `class`, and `instance`.
* Rust supports `multiprocess`, `global`, `class`, and `instance` through `abxbus::retry!`.
* Go does not include `retry` yet.
* TypeScript uses async-context re-entrancy tracking in Node/Bun to avoid same-semaphore nested deadlocks.
* `retry_on_errors` matching differs slightly:
* Python: exception classes or compiled regex patterns (matched against `"ErrorClass: message"`).
* TypeScript: error constructors, error-name strings, or regex patterns.
* Rust: typed predicate function via `retry_if = should_retry`.
# Backpressure
Source: https://abxbus.archivebox.io/concurrency/backpressure
How emit, queueing, and history limits interact under high event volume.
Backpressure in `abxbus` is history-policy based, not queue-capacity based.
* `emit()` enqueues synchronously and returns immediately.
* Pending queues are unbounded in each runtime.
* Overload behavior is controlled by `max_history_size` + `max_history_drop`.
## 1) If I emit 1,000,000 events, will errors be raised?
### Error conditions
| Runtime | Condition | What is raised |
| - | - | - |
| Python | `emit()` called with no running event loop | `RuntimeError` (`emit() called but no event loop is running`) |
| Python | `max_history_size > 0`, `max_history_drop=False`, and history already at limit | `RuntimeError` (`history limit reached`) |
| TypeScript | `emit()` with `max_history_size > 0`, `max_history_drop=false`, and history already at limit | `Error` (message contains `history limit reached`) |
| Rust | typed `emit::(...)` with `max_history_size > 0`, `max_history_drop=false`, and history already at limit | panic with `history limit reached` |
| Go | `Emit()` with `MaxHistorySize > 0`, `MaxHistoryDrop=false`, and history already at limit | panic with `history limit reached` |
| All | Process runs out of memory under extreme load | Runtime/VM OOM failure (not a bus-specific exception type) |
In normal operation, queue-capacity errors are not the backpressure mechanism.
`max_history_size=0` / `MaxHistorySize=0` is a special case in all runtimes: it does not trigger history-limit rejection, and instead keeps only in-flight visibility.
With `max_history_drop=true`, `emit()` does not reject on history size. Python and TypeScript evict completed children before parents, preserving active work and retained ancestry even when they temporarily exceed the limit.
### Reject vs drop behavior
```python theme={null}
from abxbus import EventBus
# Reject new emits once history reaches N
reject_bus = EventBus(max_history_size=10_000, max_history_drop=False)
# Never reject on history size; trim oldest history entries instead
drop_bus = EventBus(max_history_size=10_000, max_history_drop=True)
```
```ts theme={null}
import { EventBus } from 'abxbus'
const rejectBus = new EventBus('RejectBus', { max_history_size: 10_000, max_history_drop: false })
const dropBus = new EventBus('DropBus', { max_history_size: 10_000, max_history_drop: true })
```
```rust theme={null}
use abxbus::event_bus::{EventBus, EventBusOptions};
let reject_bus = EventBus::new_with_options(Some("RejectBus".to_string()), EventBusOptions {
max_history_size: Some(10_000),
max_history_drop: false,
..EventBusOptions::default()
});
let drop_bus = EventBus::new_with_options(Some("DropBus".to_string()), EventBusOptions {
max_history_size: Some(10_000),
max_history_drop: true,
..EventBusOptions::default()
});
```
```go theme={null}
maxHistory := 10_000
// Reject new emits once history reaches N.
rejectBus := abxbus.NewEventBus("RejectBus", &abxbus.EventBusOptions{
MaxHistorySize: &maxHistory,
MaxHistoryDrop: false,
})
// Never reject on history size; trim oldest history entries instead.
dropBus := abxbus.NewEventBus("DropBus", &abxbus.EventBusOptions{
MaxHistorySize: &maxHistory,
MaxHistoryDrop: true,
})
```
## 2) If 1,000,000 events complete, how many are kept?
Let `N = max_history_size`.
| Setting | Events retained after bus becomes idle | Notes |
| - | - | - |
| `N = None` / `null` | All completed events (so up to 1,000,000) | History is unbounded. |
| `N > 0`, `max_history_drop = false` | Up to `N` | New emits are rejected once history reaches `N`. |
| `N > 0`, `max_history_drop = true` | Bounded to `N` at steady state | Oldest history entries are removed first. |
| `N = 0` | `0` completed events retained | Only pending/in-flight visibility is kept; completed entries are dropped. |
Python nuance: in heavy bursts with `max_history_drop=True`, cleanup is amortized, so history can temporarily exceed `N` before converging back to `<= N`.
For the broader retention model, see [Event History Store](../features/event-history-store).
## 3) How RAM usage scales
At a high level, memory grows with:
* pending queue depth,
* retained history size,
* per-event handler/result payload size.
A practical model is:
`RAM ~= O(pending_event_queue) + O(event_history) + O(event_results and payloads)`
### Measured slopes from perf suites
* Python README matrix reports scenario-dependent peak RSS slopes between about `0.025kb/event` and `8.024kb/event`.
* TypeScript README matrix reports scenario/runtime-dependent peak RSS slopes between about `0.1kb/event` and `7.9kb/event`.
* TypeScript README notes those `kb/event` values are measured during active processing with history aggressively bounded (`max_history_size=1` in perf harnesses).
Use those numbers as throughput-era slope indicators, not exact long-term retention multipliers for your payloads.
Operationally:
* bounded history (`N` finite) keeps steady-state memory bounded by queue depth + `N`,
* unbounded history (`N=None/null`) makes retained RAM grow roughly linearly with total completed events.
## 4) Queue vs history lifecycle (exact behavior)
Events do not "move from queue to history." They are added to history at `emit()` time, and can exist in both structures while pending.
### Python timeline (`emit`)
1. Validate pressure policy.
2. Enqueue into `pending_event_queue`.
3. Add same event object to `event_history`.
4. Runloop dequeues event (`queue.get()`), then executes handlers.
5. Event remains in `event_history` as `pending` -> `started` -> `completed` unless trimmed/removed by history policy.
### TypeScript timeline (`emit`)
1. Validate pressure policy.
2. Add event to `event_history`.
3. Apply `trimHistory()`.
4. Push event into `pending_event_queue`.
5. Runloop shifts from queue and executes handlers.
6. Event remains in `event_history` unless trimmed/removed by policy.
### Go timeline (`Emit`)
1. Validate pressure policy against `EventHistory`.
2. Add event to `EventHistory`.
3. Apply history trimming according to `MaxHistorySize` + `MaxHistoryDrop`.
4. Push event into the pending queue.
5. Runloop dequeues the event and executes handlers.
6. Event remains in `EventHistory` unless trimmed/removed by policy.
### Rust timeline (`emit`)
1. Validate pressure policy against the runtime history store.
2. Add event to history and append its id to history order.
3. Apply history trimming according to `max_history_size` + `max_history_drop`.
4. Push event into the pending queue.
5. Runloop dequeues the event and executes handlers.
6. Event remains in history unless trimmed/removed by policy.
So yes:
* an event can be in both `pending_event_queue` and `event_history` at the same time,
* `event_history` can contain pending events (not only started/completed events).
## Observe both structures directly
```python theme={null}
event = bus.emit(MyEvent())
pending_count = bus.pending_event_queue.qsize() if bus.pending_event_queue else 0
history_count = len(bus.event_history)
print('pending_event_queue=', pending_count, 'event_history=', history_count)
# pending_event_queue= 1 event_history= 1
```
```ts theme={null}
const event = bus.emit(MyEvent({}))
console.log('pending_event_queue=', bus.pending_event_queue.length, 'event_history=', bus.event_history.size)
```
```rust theme={null}
let event = bus.emit(MyEvent {
..Default::default()
});
let snapshot = bus.to_json_value();
let pending_count = snapshot["pending_event_queue"].as_array().map(Vec::len).unwrap_or(0);
let history_count = snapshot["event_history"].as_object().map(|items| items.len()).unwrap_or(0);
println!("pending_event_queue={pending_count} event_history={history_count}");
```
```go theme={null}
event := bus.Emit(abxbus.NewBaseEvent("MyEvent", nil))
var state struct {
PendingEventQueue []string `json:"pending_event_queue"`
EventHistory map[string]any `json:"event_history"`
}
snapshot, err := bus.ToJSON()
if err != nil {
panic(err)
}
if err := json.Unmarshal(snapshot, &state); err != nil {
panic(err)
}
pendingCount := len(state.PendingEventQueue)
historyCount := len(state.EventHistory)
fmt.Println("pending_event_queue=", pendingCount, "event_history=", historyCount)
// pending_event_queue= 1 event_history= 1
_ = event
```
# Events: bus-serial
Source: https://abxbus.archivebox.io/concurrency/events-bus-serial
Process one event at a time per bus, while allowing overlap across buses.
`bus-serial` enforces one active event per bus, while different buses can process events simultaneously.
Companion runnable example:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
## Lifecycle impact
1. Events enqueue per bus in FIFO order.
2. Each bus holds its own event lock.
3. A busy bus does not block other buses.
4. Queue-jump child events can preempt that same bus queue when awaited in-handler.
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class WorkEvent(BaseEvent):
order: int
source: str
bus_a = EventBus('BusSerialA', event_concurrency='bus-serial')
bus_b = EventBus('BusSerialB', event_concurrency='bus-serial')
starts_a: list[int] = []
starts_b: list[int] = []
in_flight_global = 0
max_in_flight_global = 0
async def handler_a(event: WorkEvent) -> None:
global in_flight_global, max_in_flight_global
in_flight_global += 1
max_in_flight_global = max(max_in_flight_global, in_flight_global)
starts_a.append(event.order)
await asyncio.sleep(0.01)
in_flight_global -= 1
async def handler_b(event: WorkEvent) -> None:
global in_flight_global, max_in_flight_global
in_flight_global += 1
max_in_flight_global = max(max_in_flight_global, in_flight_global)
starts_b.append(event.order)
await asyncio.sleep(0.01)
in_flight_global -= 1
bus_a.on(WorkEvent, handler_a)
bus_b.on(WorkEvent, handler_b)
for i in range(4):
bus_a.emit(WorkEvent(order=i, source='a'))
bus_b.emit(WorkEvent(order=i, source='b'))
await bus_a.wait_until_idle()
await bus_b.wait_until_idle()
assert starts_a == [0, 1, 2, 3]
assert starts_b == [0, 1, 2, 3]
assert max_in_flight_global >= 2
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const WorkEvent = BaseEvent.extend('WorkEvent', {
order: z.number(),
source: z.string(),
})
const busA = new EventBus('BusSerialA', { event_concurrency: 'bus-serial' })
const busB = new EventBus('BusSerialB', { event_concurrency: 'bus-serial' })
const startsA: number[] = []
const startsB: number[] = []
busA.on(WorkEvent, async (event) => {
startsA.push(event.order)
await new Promise((resolve) => setTimeout(resolve, 2))
})
busB.on(WorkEvent, async (event) => {
startsB.push(event.order)
await new Promise((resolve) => setTimeout(resolve, 2))
})
for (let i = 0; i < 4; i += 1) {
busA.emit(WorkEvent({ order: i, source: 'a' }))
busB.emit(WorkEvent({ order: i, source: 'b' }))
}
await busA.waitUntilIdle()
await busB.waitUntilIdle()
if (JSON.stringify(startsA) !== JSON.stringify([0, 1, 2, 3])) throw new Error('bus A FIFO failed')
if (JSON.stringify(startsB) !== JSON.stringify([0, 1, 2, 3])) throw new Error('bus B FIFO failed')
```
```go theme={null}
package main
import (
"reflect"
"sync"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
busA := abxbus.NewEventBus("BusSerialA", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyBusSerial,
})
busB := abxbus.NewEventBus("BusSerialB", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyBusSerial,
})
var mu sync.Mutex
startsA := []int{}
startsB := []int{}
inFlightGlobal := 0
maxInFlightGlobal := 0
handler := func(starts *[]int) func(*abxbus.BaseEvent) (any, error) {
return func(event *abxbus.BaseEvent) (any, error) {
mu.Lock()
inFlightGlobal++
if inFlightGlobal > maxInFlightGlobal {
maxInFlightGlobal = inFlightGlobal
}
*starts = append(*starts, event.Payload["order"].(int))
mu.Unlock()
time.Sleep(10 * time.Millisecond)
mu.Lock()
inFlightGlobal--
mu.Unlock()
return nil, nil
}
}
busA.On("WorkEvent", "handler_a", handler(&startsA), nil)
busB.On("WorkEvent", "handler_b", handler(&startsB), nil)
for i := 0; i < 4; i++ {
busA.Emit(abxbus.NewBaseEvent("WorkEvent", map[string]any{"order": i, "source": "a"}))
busB.Emit(abxbus.NewBaseEvent("WorkEvent", map[string]any{"order": i, "source": "b"}))
}
busA.WaitUntilIdle(nil)
busB.WaitUntilIdle(nil)
if !reflect.DeepEqual(startsA, []int{0, 1, 2, 3}) {
panic("bus A FIFO failed")
}
if !reflect.DeepEqual(startsB, []int{0, 1, 2, 3}) {
panic("bus B FIFO failed")
}
if maxInFlightGlobal < 2 {
panic("expected buses to overlap")
}
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::EventConcurrencyMode,
};
use futures::executor::block_on;
event! {
struct WorkEvent {
event_result_type: (),
}
}
let bus_a = EventBus::new_with_options(
Some("BusSerialA".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::BusSerial,
..EventBusOptions::default()
},
);
let bus_b = EventBus::new_with_options(
Some("BusSerialB".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::BusSerial,
..EventBusOptions::default()
},
);
bus_a.on(WorkEvent, |_event: WorkEvent| async move { Ok(()) });
bus_b.on(WorkEvent, |_event: WorkEvent| async move { Ok(()) });
for _ in 0..4 {
bus_a.emit(WorkEvent { ..Default::default() });
bus_b.emit(WorkEvent { ..Default::default() });
}
block_on(bus_a.wait_until_idle(None));
block_on(bus_b.wait_until_idle(None));
```
## Notes
* This is typically the best default for multi-bus systems.
* It preserves local determinism while retaining cross-bus throughput.
# Events: global-serial
Source: https://abxbus.archivebox.io/concurrency/events-global-serial
Process only one event at a time across all buses.
`global-serial` enforces a single global event-processing slot across all `EventBus` instances.
Note that the global lock is scoped to the `EventBus` class; if you need separate global lock domains, subclass `EventBus`.
Companion runnable example:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
## Lifecycle impact
1. An emitted event is queued on its target bus as usual.
2. Before handler execution starts, the bus acquires the shared global event lock.
3. While one event is running anywhere, other buses wait.
4. Handler-level concurrency still applies inside that one active event.
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class SerialEvent(BaseEvent):
order: int
source: str
bus_a = EventBus('GlobalSerialA', event_concurrency='global-serial')
bus_b = EventBus('GlobalSerialB', event_concurrency='global-serial')
in_flight = 0
max_in_flight = 0
starts: list[str] = []
async def handler(event: SerialEvent) -> None:
global in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
starts.append(f'{event.source}:{event.order}')
await asyncio.sleep(0.01)
in_flight -= 1
bus_a.on(SerialEvent, handler)
bus_b.on(SerialEvent, handler)
for i in range(3):
bus_a.emit(SerialEvent(order=i, source='a'))
bus_b.emit(SerialEvent(order=i, source='b'))
await bus_a.wait_until_idle()
await bus_b.wait_until_idle()
assert max_in_flight == 1
assert [s for s in starts if s.startswith('a:')] == ['a:0', 'a:1', 'a:2']
assert [s for s in starts if s.startswith('b:')] == ['b:0', 'b:1', 'b:2']
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const SerialEvent = BaseEvent.extend('SerialEvent', {
order: z.number(),
source: z.string(),
})
const busA = new EventBus('GlobalSerialA', { event_concurrency: 'global-serial' })
const busB = new EventBus('GlobalSerialB', { event_concurrency: 'global-serial' })
let inFlight = 0
let maxInFlight = 0
const starts: string[] = []
const handler = async (event: InstanceType) => {
inFlight += 1
maxInFlight = Math.max(maxInFlight, inFlight)
starts.push(`${event.source}:${event.order}`)
await new Promise((resolve) => setTimeout(resolve, 10))
inFlight -= 1
}
busA.on(SerialEvent, handler)
busB.on(SerialEvent, handler)
for (let i = 0; i < 3; i += 1) {
busA.emit(SerialEvent({ order: i, source: 'a' }))
busB.emit(SerialEvent({ order: i, source: 'b' }))
}
await busA.waitUntilIdle()
await busB.waitUntilIdle()
if (maxInFlight !== 1) throw new Error('expected global serialization')
```
```go theme={null}
package main
import (
"strings"
"sync"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
options := &abxbus.EventBusOptions{EventConcurrency: abxbus.EventConcurrencyGlobalSerial}
busA := abxbus.NewEventBus("GlobalSerialA", options)
busB := abxbus.NewEventBus("GlobalSerialB", options)
var mu sync.Mutex
inFlight := 0
maxInFlight := 0
starts := []string{}
handler := func(event *abxbus.BaseEvent) (any, error) {
mu.Lock()
inFlight++
if inFlight > maxInFlight {
maxInFlight = inFlight
}
starts = append(starts, event.Payload["source"].(string))
mu.Unlock()
time.Sleep(10 * time.Millisecond)
mu.Lock()
inFlight--
mu.Unlock()
return nil, nil
}
busA.On("SerialEvent", "handler", handler, nil)
busB.On("SerialEvent", "handler", handler, nil)
for i := 0; i < 3; i++ {
busA.Emit(abxbus.NewBaseEvent("SerialEvent", map[string]any{"order": i, "source": "a"}))
busB.Emit(abxbus.NewBaseEvent("SerialEvent", map[string]any{"order": i, "source": "b"}))
}
busA.WaitUntilIdle(nil)
busB.WaitUntilIdle(nil)
if maxInFlight != 1 {
panic("expected global serialization")
}
if strings.Count(strings.Join(starts, ""), "a") != 3 || strings.Count(strings.Join(starts, ""), "b") != 3 {
panic("expected both buses to process their queues")
}
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::EventConcurrencyMode,
};
use futures::executor::block_on;
event! {
struct SerialEvent {
event_result_type: (),
}
}
let options = EventBusOptions {
event_concurrency: EventConcurrencyMode::GlobalSerial,
..EventBusOptions::default()
};
let bus_a = EventBus::new_with_options(Some("GlobalSerialA".to_string()), options.clone());
let bus_b = EventBus::new_with_options(Some("GlobalSerialB".to_string()), options);
bus_a.on(SerialEvent, |_event: SerialEvent| async move { Ok(()) });
bus_b.on(SerialEvent, |_event: SerialEvent| async move { Ok(()) });
for _ in 0..3 {
bus_a.emit(SerialEvent { ..Default::default() });
bus_b.emit(SerialEvent { ..Default::default() });
}
block_on(bus_a.wait_until_idle(None));
block_on(bus_b.wait_until_idle(None));
```
## Notes
* This mode is strongest for determinism across distributed in-process bus topologies.
* Queue-jump behavior (`await event.now()` inside handlers) still applies, but it does so under the same global lock.
# Events: parallel
Source: https://abxbus.archivebox.io/concurrency/events-parallel
Allow multiple events to execute concurrently on the same bus.
`parallel` removes event-level serialization for a bus, so multiple events can be in-flight simultaneously.
Companion runnable example:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
## Lifecycle impact
1. Events still enqueue and are tracked in history.
2. The bus does not gate execution with an event semaphore.
3. Handler-level concurrency rules still apply within each event.
4. Ordering guarantees become weaker under load because events can overlap.
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class ParallelEvent(BaseEvent):
order: int
bus = EventBus('ParallelEventBus', event_concurrency='parallel', event_handler_concurrency='parallel')
in_flight = 0
max_in_flight = 0
release = asyncio.Event()
async def handler(_: ParallelEvent) -> None:
global in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
await release.wait()
await asyncio.sleep(0.01)
in_flight -= 1
bus.on(ParallelEvent, handler)
bus.emit(ParallelEvent(order=0))
bus.emit(ParallelEvent(order=1))
await asyncio.sleep(0)
release.set()
await bus.wait_until_idle()
assert max_in_flight >= 2
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const ParallelEvent = BaseEvent.extend('ParallelEvent', { order: z.number() })
const bus = new EventBus('ParallelEventBus', {
event_concurrency: 'parallel',
event_handler_concurrency: 'parallel',
})
let inFlight = 0
let maxInFlight = 0
let release!: () => void
const gate = new Promise((resolve) => {
release = resolve
})
bus.on(ParallelEvent, async () => {
inFlight += 1
maxInFlight = Math.max(maxInFlight, inFlight)
await gate
await new Promise((resolve) => setTimeout(resolve, 10))
inFlight -= 1
})
bus.emit(ParallelEvent({ order: 0 }))
bus.emit(ParallelEvent({ order: 1 }))
await new Promise((resolve) => setTimeout(resolve, 0))
release()
await bus.waitUntilIdle()
if (maxInFlight < 2) throw new Error('expected overlapping events')
```
```go theme={null}
package main
import (
"sync"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("ParallelEventBus", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyParallel,
EventHandlerConcurrency: abxbus.EventHandlerConcurrencyParallel,
})
var mu sync.Mutex
inFlight := 0
maxInFlight := 0
release := make(chan struct{})
bus.On("ParallelEvent", "handler", func(event *abxbus.BaseEvent) (any, error) {
mu.Lock()
inFlight++
if inFlight > maxInFlight {
maxInFlight = inFlight
}
mu.Unlock()
<-release
time.Sleep(10 * time.Millisecond)
mu.Lock()
inFlight--
mu.Unlock()
return nil, nil
}, nil)
bus.Emit(abxbus.NewBaseEvent("ParallelEvent", map[string]any{"order": 0}))
bus.Emit(abxbus.NewBaseEvent("ParallelEvent", map[string]any{"order": 1}))
time.Sleep(10 * time.Millisecond)
close(release)
bus.WaitUntilIdle(nil)
if maxInFlight < 2 {
panic("expected overlapping events")
}
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::{EventConcurrencyMode, EventHandlerConcurrencyMode},
};
use futures::executor::block_on;
event! {
struct ParallelEvent {
event_result_type: (),
}
}
let bus = EventBus::new_with_options(
Some("ParallelEventBus".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::Parallel,
event_handler_concurrency: EventHandlerConcurrencyMode::Parallel,
..EventBusOptions::default()
},
);
bus.on(ParallelEvent, |_event: ParallelEvent| async move { Ok(()) });
bus.emit(ParallelEvent { ..Default::default() });
bus.emit(ParallelEvent { ..Default::default() });
block_on(bus.wait_until_idle(None));
```
## Notes
* Use when throughput matters more than deterministic event ordering.
* Combine with idempotent handlers and explicit external coordination when needed.
# Handler Completion: all
Source: https://abxbus.archivebox.io/concurrency/handler-completion-all
Wait for every matching handler before an event completes.
`all` is the default handler completion mode. The event completes only after every matching handler reaches a terminal state.
## Lifecycle impact
1. All matching handlers are allowed to run.
2. A successful early handler does not short-circuit the event.
3. Event completion waits for every handler to finish, fail, or time out.
4. Result collection includes all successful non-`None` / non-`undefined` return values.
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class CompletionEvent(BaseEvent[str]):
pass
bus = EventBus(
'CompletionAllBus',
event_handler_concurrency='parallel',
event_handler_completion='all',
)
seen: list[str] = []
async def fast_handler(_: CompletionEvent) -> str:
await asyncio.sleep(0.01)
seen.append('fast')
return 'fast'
async def slow_handler(_: CompletionEvent) -> str:
await asyncio.sleep(0.05)
seen.append('slow')
return 'slow'
bus.on(CompletionEvent, fast_handler)
bus.on(CompletionEvent, slow_handler)
event = bus.emit(CompletionEvent())
await event.now()
assert set(seen) == {'fast', 'slow'}
results = await event.event_results_list(raise_if_any=False, raise_if_none=False)
assert set(results) == {'fast', 'slow'}
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const CompletionEvent = BaseEvent.extend('CompletionEvent', { event_result_type: z.string() })
const bus = new EventBus('CompletionAllBus', {
event_handler_concurrency: 'parallel',
event_handler_completion: 'all',
})
const seen: string[] = []
const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
bus.on(CompletionEvent, async () => {
await delay(10)
seen.push('fast')
return 'fast'
})
bus.on(CompletionEvent, async () => {
await delay(50)
seen.push('slow')
return 'slow'
})
const event = bus.emit(CompletionEvent({}))
await event.now()
const results = await event.eventResultsList({ raise_if_any: false, raise_if_none: false })
if (seen.length !== 2) throw new Error('expected all handlers to run')
if (!results.includes('fast') || !results.includes('slow')) {
throw new Error('expected both handler return values')
}
```
```go theme={null}
package main
import (
"sync"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("CompletionAllBus", &abxbus.EventBusOptions{
EventHandlerConcurrency: abxbus.EventHandlerConcurrencyParallel,
EventHandlerCompletion: abxbus.EventHandlerCompletionAll,
})
var mu sync.Mutex
seen := map[string]bool{}
bus.On("CompletionEvent", "fast_handler", func(event *abxbus.BaseEvent) (any, error) {
time.Sleep(10 * time.Millisecond)
mu.Lock()
seen["fast"] = true
mu.Unlock()
return "fast", nil
}, nil)
bus.On("CompletionEvent", "slow_handler", func(event *abxbus.BaseEvent) (any, error) {
time.Sleep(50 * time.Millisecond)
mu.Lock()
seen["slow"] = true
mu.Unlock()
return "slow", nil
}, nil)
event := bus.Emit(abxbus.NewBaseEvent("CompletionEvent", nil))
if _, err := event.Now(); err != nil {
panic(err)
}
results, err := event.EventResultsList(
&abxbus.EventResultOptions{RaiseIfAny: false, RaiseIfNone: false},
)
if err != nil {
panic(err)
}
mu.Lock()
allSeen := seen["fast"] && seen["slow"]
mu.Unlock()
if !allSeen {
panic("expected all handlers to run")
}
if len(results) != 2 {
panic("expected both handler return values")
}
}
```
```rust theme={null}
use abxbus::{
base_event::EventResultOptions,
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::{EventHandlerCompletionMode, EventHandlerConcurrencyMode},
};
use futures::executor::block_on;
use serde_json::json;
event! {
struct CompletionEvent {
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new_with_options(
Some("CompletionAllBus".to_string()),
EventBusOptions {
event_handler_concurrency: EventHandlerConcurrencyMode::Parallel,
event_handler_completion: EventHandlerCompletionMode::All,
..EventBusOptions::default()
},
);
bus.on(CompletionEvent, |_event: CompletionEvent| async move { Ok(json!("fast")) });
bus.on(CompletionEvent, |_event: CompletionEvent| async move { Ok(json!("slow")) });
let event = bus.emit(CompletionEvent { ..Default::default() });
let results = block_on(event.event_results_list_with_options(EventResultOptions {
raise_if_any: false,
raise_if_none: false,
..EventResultOptions::default()
}))?;
assert_eq!(results.len(), 2);
```
## Notes
* `all` is best when multiple handlers contribute required side effects.
* Handler scheduling (`serial` vs `parallel`) changes overlap, but not the fact that all handlers must settle.
# Handler Completion: first
Source: https://abxbus.archivebox.io/concurrency/handler-completion-first
Complete an event on the first successful handler result.
`first` short-circuits event completion once the first successful non-`None` / non-`undefined` result is available.
## Lifecycle impact
1. The first successful result wins (`None`/`undefined` and errors do not win).
2. In `serial` handler mode, remaining handlers are skipped once a winner appears.
3. In `parallel` handler mode, in-flight losers are cancelled or aborted.
4. Event completion resolves as soon as a winner is found (or all handlers fail).
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class CompletionEvent(BaseEvent[str]):
pass
bus = EventBus(
'CompletionFirstBus',
event_handler_concurrency='parallel',
event_handler_completion='first',
)
state = {'slow_started': False, 'slow_cancelled': False}
async def fast_handler(_: CompletionEvent) -> str:
await asyncio.sleep(0.01)
return 'winner'
async def slow_handler(_: CompletionEvent) -> str:
state['slow_started'] = True
try:
await asyncio.sleep(0.5)
return 'slow'
except asyncio.CancelledError:
state['slow_cancelled'] = True
raise
bus.on(CompletionEvent, slow_handler)
bus.on(CompletionEvent, fast_handler)
event = bus.emit(CompletionEvent())
await event.now()
value = await event.event_result(raise_if_any=False, raise_if_none=False)
assert value == 'winner'
assert state['slow_started'] is True
assert state['slow_cancelled'] is True
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const CompletionEvent = BaseEvent.extend('CompletionEvent', { event_result_type: z.string() })
const bus = new EventBus('CompletionFirstBus', {
event_handler_concurrency: 'parallel',
event_handler_completion: 'first',
})
let slowStarted = false
let slowCompleted = false
const delay = (ms: number) => new Promise((resolve) => setTimeout(resolve, ms))
bus.on(CompletionEvent, async () => {
slowStarted = true
await delay(500)
slowCompleted = true
return 'slow'
})
bus.on(CompletionEvent, async () => {
await delay(10)
return 'winner'
})
const event = bus.emit(CompletionEvent({}))
await event.now()
if ((await event.eventResult({ raise_if_any: false, raise_if_none: false })) !== 'winner') {
throw new Error('expected first winner result')
}
if (!slowStarted) throw new Error('expected slow handler to start')
if (slowCompleted) throw new Error('slow handler should not complete before event resolves')
```
```go theme={null}
package main
import (
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("CompletionFirstBus", &abxbus.EventBusOptions{
EventHandlerConcurrency: abxbus.EventHandlerConcurrencyParallel,
EventHandlerCompletion: abxbus.EventHandlerCompletionFirst,
})
slowStarted := make(chan struct{}, 1)
slowCompleted := make(chan struct{}, 1)
bus.On("CompletionEvent", "slow_handler", func(event *abxbus.BaseEvent) (any, error) {
slowStarted <- struct{}{}
time.Sleep(500 * time.Millisecond)
slowCompleted <- struct{}{}
return "slow", nil
}, nil)
bus.On("CompletionEvent", "fast_handler", func(event *abxbus.BaseEvent) (any, error) {
time.Sleep(10 * time.Millisecond)
return "winner", nil
}, nil)
event := bus.Emit(abxbus.NewBaseEvent("CompletionEvent", nil))
_, err := event.Now(&abxbus.EventWaitOptions{FirstResult: true})
if err != nil {
panic(err)
}
value, err := event.EventResult(&abxbus.EventResultOptions{
RaiseIfAny: false,
RaiseIfNone: false,
})
if err != nil {
panic(err)
}
if value != "winner" {
panic("expected first winner result")
}
select {
case <-slowStarted:
default:
panic("expected slow handler to start")
}
select {
case <-slowCompleted:
panic("slow handler should not complete before event resolves")
default:
}
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::{EventHandlerCompletionMode, EventHandlerConcurrencyMode},
};
use futures::executor::block_on;
use futures_timer::Delay;
use serde_json::json;
use std::time::Duration;
event! {
struct CompletionEvent {
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new_with_options(
Some("CompletionFirstBus".to_string()),
EventBusOptions {
event_handler_concurrency: EventHandlerConcurrencyMode::Parallel,
event_handler_completion: EventHandlerCompletionMode::First,
..EventBusOptions::default()
},
);
bus.on(CompletionEvent, |_event: CompletionEvent| async move {
Delay::new(Duration::from_millis(500)).await;
Ok(json!("slow"))
});
bus.on(CompletionEvent, |_event: CompletionEvent| async move {
Delay::new(Duration::from_millis(10)).await;
Ok(json!("winner"))
});
let event = bus.emit(CompletionEvent { ..Default::default() });
let value = block_on(event.event_result())?;
assert_eq!(value, Some(json!("winner")));
```
## Notes
* This mode is useful for fallback chains and race-to-first-response patterns.
* Use `await event.now(first_result=True)` / `await event.now({ first_result: true })` / `event.Now(&abxbus.EventWaitOptions{FirstResult: true})` / `event.now_with_options(EventWaitOptions { first_result: true, ..Default::default() })` when you want to resolve as soon as a valid result is available while still allowing all handlers to continue. To stop running after the first valid handler result, set `event_handler_completion: 'first'` on the event definition.
# Handlers: parallel
Source: https://abxbus.archivebox.io/concurrency/handlers-parallel
Run handlers for one event concurrently.
`parallel` allows multiple handlers for the same event to run at the same time.
Companion runnable example:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
## Lifecycle impact
1. Event starts processing.
2. All applicable handlers are scheduled concurrently.
3. Event completion waits based on completion mode (`all` or `first`).
4. Per-handler timeout/error behavior remains independent per handler.
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class HandlerEvent(BaseEvent):
pass
bus = EventBus('ParallelHandlerBus', event_handler_concurrency='parallel')
in_flight = 0
max_in_flight = 0
release = asyncio.Event()
async def tracked(_: HandlerEvent) -> None:
global in_flight, max_in_flight
in_flight += 1
max_in_flight = max(max_in_flight, in_flight)
await release.wait()
in_flight -= 1
bus.on(HandlerEvent, tracked)
bus.on(HandlerEvent, tracked)
event = bus.emit(HandlerEvent())
await asyncio.sleep(0)
release.set()
await event.now()
assert max_in_flight >= 2
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
const HandlerEvent = BaseEvent.extend('HandlerEvent', {})
const bus = new EventBus('ParallelHandlerBus', { event_handler_concurrency: 'parallel' })
let inFlight = 0
let maxInFlight = 0
let release!: () => void
const gate = new Promise((resolve) => {
release = resolve
})
const tracked = async () => {
inFlight += 1
maxInFlight = Math.max(maxInFlight, inFlight)
await gate
inFlight -= 1
}
bus.on(HandlerEvent, tracked)
bus.on(HandlerEvent, tracked)
const event = bus.emit(HandlerEvent({}))
await new Promise((resolve) => setTimeout(resolve, 0))
release()
await event.now()
if (maxInFlight < 2) throw new Error('expected overlapping handlers')
```
```go theme={null}
package main
import (
"sync"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("ParallelHandlerBus", &abxbus.EventBusOptions{
EventHandlerConcurrency: abxbus.EventHandlerConcurrencyParallel,
})
var mu sync.Mutex
inFlight := 0
maxInFlight := 0
started := make(chan struct{}, 2)
release := make(chan struct{})
tracked := func(event *abxbus.BaseEvent) (any, error) {
mu.Lock()
inFlight++
if inFlight > maxInFlight {
maxInFlight = inFlight
}
mu.Unlock()
started <- struct{}{}
<-release
mu.Lock()
inFlight--
mu.Unlock()
return nil, nil
}
bus.On("HandlerEvent", "tracked_a", tracked, nil)
bus.On("HandlerEvent", "tracked_b", tracked, nil)
event := bus.Emit(abxbus.NewBaseEvent("HandlerEvent", nil))
<-started
<-started
close(release)
if _, err := event.Now(); err != nil {
panic(err)
}
if maxInFlight < 2 {
panic("expected overlapping handlers")
}
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::EventHandlerConcurrencyMode,
};
use futures::executor::block_on;
event! {
struct HandlerEvent {
event_result_type: (),
}
}
let bus = EventBus::new_with_options(
Some("ParallelHandlerBus".to_string()),
EventBusOptions {
event_handler_concurrency: EventHandlerConcurrencyMode::Parallel,
..EventBusOptions::default()
},
);
bus.on(HandlerEvent, |_event: HandlerEvent| async move { Ok(()) });
bus.on(HandlerEvent, |_event: HandlerEvent| async move { Ok(()) });
let event = bus.emit(HandlerEvent { ..Default::default() });
block_on(event.now());
```
## Notes
* Best for independent I/O-bound handlers where overlap reduces total latency.
* If handlers mutate shared resources, add explicit synchronization.
# Handlers: serial
Source: https://abxbus.archivebox.io/concurrency/handlers-serial
Run handlers one at a time per event, in registration order.
`serial` executes handlers for a single event sequentially.
Companion runnable example:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
## Lifecycle impact
1. Event starts processing.
2. Handler A runs to completion (or failure/timeout).
3. Handler B starts afterward, then C, and so on.
4. Event completion waits for the serial chain (or completion-mode short-circuit rules).
## Execution order example
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class HandlerEvent(BaseEvent):
pass
bus = EventBus('SerialHandlerBus', event_handler_concurrency='serial')
log: list[str] = []
async def h1(_: HandlerEvent) -> None:
log.append('h1_start')
await asyncio.sleep(0.01)
log.append('h1_end')
async def h2(_: HandlerEvent) -> None:
log.append('h2_start')
await asyncio.sleep(0.01)
log.append('h2_end')
bus.on(HandlerEvent, h1)
bus.on(HandlerEvent, h2)
await bus.emit(HandlerEvent()).now()
assert log == ['h1_start', 'h1_end', 'h2_start', 'h2_end']
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
const HandlerEvent = BaseEvent.extend('HandlerEvent', {})
const bus = new EventBus('SerialHandlerBus', { event_handler_concurrency: 'serial' })
const log: string[] = []
bus.on(HandlerEvent, async () => {
log.push('h1_start')
await new Promise((resolve) => setTimeout(resolve, 10))
log.push('h1_end')
})
bus.on(HandlerEvent, async () => {
log.push('h2_start')
await new Promise((resolve) => setTimeout(resolve, 10))
log.push('h2_end')
})
await bus.emit(HandlerEvent({})).now()
if (JSON.stringify(log) !== JSON.stringify(['h1_start', 'h1_end', 'h2_start', 'h2_end'])) {
throw new Error('expected serial handler execution order')
}
```
```go theme={null}
package main
import (
"reflect"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("SerialHandlerBus", &abxbus.EventBusOptions{
EventHandlerConcurrency: abxbus.EventHandlerConcurrencySerial,
})
log := []string{}
bus.On("HandlerEvent", "h1", func(event *abxbus.BaseEvent) (any, error) {
log = append(log, "h1_start")
time.Sleep(10 * time.Millisecond)
log = append(log, "h1_end")
return nil, nil
}, nil)
bus.On("HandlerEvent", "h2", func(event *abxbus.BaseEvent) (any, error) {
log = append(log, "h2_start")
time.Sleep(10 * time.Millisecond)
log = append(log, "h2_end")
return nil, nil
}, nil)
event := bus.Emit(abxbus.NewBaseEvent("HandlerEvent", nil))
if _, err := event.Now(); err != nil {
panic(err)
}
expected := []string{"h1_start", "h1_end", "h2_start", "h2_end"}
if !reflect.DeepEqual(log, expected) {
panic("expected serial handler execution order")
}
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
BaseEvent,
types::EventHandlerConcurrencyMode,
};
use futures::executor::block_on;
event! {
struct HandlerEvent {
event_result_type: (),
}
}
let bus = EventBus::new_with_options(
Some("SerialHandlerBus".to_string()),
EventBusOptions {
event_handler_concurrency: EventHandlerConcurrencyMode::Serial,
..EventBusOptions::default()
},
);
bus.on(HandlerEvent, |_event: HandlerEvent| async move { Ok(()) });
bus.on(HandlerEvent, |_event: HandlerEvent| async move { Ok(()) });
let event = bus.emit(HandlerEvent { ..Default::default() });
block_on(event.now());
```
## Notes
* Best when handlers share mutable state or require strict ordering.
* Execution remains predictable but may increase per-event latency.
# Immediate Execution (RPC-style)
Source: https://abxbus.archivebox.io/concurrency/immediate-execution
Queue-jump behavior for awaiting child events inside handlers.
Immediate execution lets a handler emit a child event and await it like a direct async function call.
When this happens inside a handler, the child event is processed immediately (queue-jump) instead of waiting behind unrelated queued events.
Repository example files:
* [`examples/immediate_event_processing.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/immediate_event_processing.py)
* [`abxbus-ts/examples/immediate_event_processing.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/immediate_event_processing.ts)
* [`abxbus-rust/tests/test_eventbus_timeout.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_eventbus_timeout.rs)
* [`abxbus-go/tests/eventbus_dispatch_parent_tracking_test.go`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-go/tests/eventbus_dispatch_parent_tracking_test.go)
## Core pattern
```python theme={null}
from abxbus import BaseEvent, EventBus
class ParentEvent(BaseEvent[str]):
pass
class ChildEvent(BaseEvent[str]):
pass
bus = EventBus('RpcBus')
async def on_parent(event: ParentEvent) -> str:
child = event.emit(ChildEvent())
await child.now() # queue-jump while still inside this handler
value = await child.event_result()
return f'parent got: {value}'
async def on_child(_: ChildEvent) -> str:
return 'child response'
bus.on(ParentEvent, on_parent)
bus.on(ChildEvent, on_child)
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const ParentEvent = BaseEvent.extend('ParentEvent', { event_result_type: z.string() })
const ChildEvent = BaseEvent.extend('ChildEvent', { event_result_type: z.string() })
const bus = new EventBus('RpcBus')
bus.on(ParentEvent, async (event) => {
const child = event.emit(ChildEvent({}))
await child.now() // queue-jump while still inside this handler
return `parent got: ${await child.eventResult()}`
})
bus.on(ChildEvent, async () => 'child response')
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use serde_json::json;
event! {
struct ParentEvent {
event_result_type: serde_json::Value,
}
}
event! {
struct ChildEvent {
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("RpcBus".to_string()));
bus.on(ParentEvent, |event: ParentEvent| async move {
let child = event.emit(ChildEvent { ..Default::default() });
child.now().await?; // queue-jump while still inside this handler
let value = child.event_result().await?;
Ok(json!(format!("parent got: {value}")))
});
bus.on(ChildEvent, |_event: ChildEvent| async move {
Ok(json!("child response"))
});
```
```go theme={null}
bus := abxbus.NewEventBus("RpcBus", nil)
bus.On("ParentEvent", "on_parent", func(event *abxbus.BaseEvent) (any, error) {
child := event.Emit(abxbus.NewBaseEvent("ChildEvent", nil))
if _, err := child.Now(); err != nil {
return nil, err
}
value, err := child.EventResult()
if err != nil {
return nil, err
}
return fmt.Sprintf("parent got: %v", value), nil
}, nil)
bus.On("ChildEvent", "on_child", func(event *abxbus.BaseEvent) (any, error) {
return "child response", nil
}, nil)
```
## Parallel fan-out inside a handler
If the parent bus/event uses `event_concurrency='parallel'`, you can queue-jump multiple child calls at once and wait for them as a group.
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class ParentEvent(BaseEvent[None]):
pass
class SomeChildEvent1(BaseEvent[str]):
pass
class SomeChildEvent2(BaseEvent[str]):
pass
class SomeChildEvent3(BaseEvent[str]):
pass
bus = EventBus('ParallelRpcBus', event_concurrency='parallel')
async def on_parent(event: ParentEvent) -> None:
settled = await asyncio.gather(
event.emit(SomeChildEvent1()),
event.emit(SomeChildEvent2()),
event.emit(SomeChildEvent3()),
return_exceptions=True,
)
for item in settled:
if isinstance(item, Exception):
print(f'child failed: {item}')
else:
print(f'child completed: {item.event_type}')
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const ParentEvent = BaseEvent.extend('ParentEvent', {})
const SomeChildEvent1 = BaseEvent.extend('SomeChildEvent1', { event_result_type: z.string() })
const SomeChildEvent2 = BaseEvent.extend('SomeChildEvent2', { event_result_type: z.string() })
const SomeChildEvent3 = BaseEvent.extend('SomeChildEvent3', { event_result_type: z.string() })
const bus = new EventBus('ParallelRpcBus', { event_concurrency: 'parallel' })
bus.on(ParentEvent, async (event) => {
const settled = await Promise.allSettled([
event.emit(SomeChildEvent1({})).now(),
event.emit(SomeChildEvent2({})).now(),
event.emit(SomeChildEvent3({})).now(),
])
for (const result of settled) {
if (result.status === 'rejected') {
console.error('child failed:', result.reason)
} else {
console.log('child completed:', result.value.event_type)
}
}
})
```
```go theme={null}
bus := abxbus.NewEventBus("ParallelRpcBus", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyParallel,
})
bus.On("ParentEvent", "on_parent", func(event *abxbus.BaseEvent) (any, error) {
children := []*abxbus.BaseEvent{
event.Emit(abxbus.NewBaseEvent("SomeChildEvent1", nil)),
event.Emit(abxbus.NewBaseEvent("SomeChildEvent2", nil)),
event.Emit(abxbus.NewBaseEvent("SomeChildEvent3", nil)),
}
for _, child := range children {
if _, err := child.Now(); err != nil {
fmt.Printf("child failed: %v\n", err)
continue
}
fmt.Printf("child completed: %s\n", child.EventType)
}
return nil, nil
}, nil)
```
```rust theme={null}
use abxbus::{event, event_bus::{EventBus, EventBusOptions}, types::EventConcurrencyMode};
use futures::future::join_all;
use serde_json::json;
event! { struct ParentEvent { event_result_type: serde_json::Value } }
event! { struct SomeChildEvent1 { event_result_type: serde_json::Value } }
event! { struct SomeChildEvent2 { event_result_type: serde_json::Value } }
event! { struct SomeChildEvent3 { event_result_type: serde_json::Value } }
let bus = EventBus::new_with_options(
Some("ParallelRpcBus".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::Parallel,
..EventBusOptions::default()
},
);
bus.on(ParentEvent, move |event: ParentEvent| {
async move {
let children = vec![
event.emit(SomeChildEvent1 { ..Default::default() }),
event.emit(SomeChildEvent2 { ..Default::default() }),
event.emit(SomeChildEvent3 { ..Default::default() }),
];
for result in join_all(children.iter().map(|child| child.now())).await {
match result {
Ok(child) => println!("child completed: {}", child.event_type),
Err(err) => println!("child failed: {err}"),
}
}
Ok(json!(null))
}
});
```
Python note: `asyncio.gather(..., return_exceptions=True)` is the closest `Promise.allSettled(...)` equivalent here. In Rust, call `now().await` on each emitted child to take the same immediate queue-jump path. In Go, call `Now()` on each emitted child.
## Execution order example
In this pattern, sibling work can already be queued, but the awaited child still runs first.
```python theme={null}
from abxbus import BaseEvent, EventBus
class ParentEvent(BaseEvent):
pass
class ChildEvent(BaseEvent):
pass
class SiblingEvent(BaseEvent):
pass
bus = EventBus('OrderBus', event_concurrency='bus-serial', event_handler_concurrency='serial')
order: list[str] = []
async def on_parent(event: ParentEvent) -> None:
order.append('parent_start')
# Detached top-level work: no parent link is recorded for the sibling.
bus.emit(SiblingEvent())
child = event.emit(ChildEvent())
await child.now()
order.append('parent_end')
async def on_child(_: ChildEvent) -> None:
order.append('child')
async def on_sibling(_: SiblingEvent) -> None:
order.append('sibling')
bus.on(ParentEvent, on_parent)
bus.on(ChildEvent, on_child)
bus.on(SiblingEvent, on_sibling)
await bus.emit(ParentEvent()).now()
await bus.wait_until_idle()
assert order.index('child') < order.index('parent_end')
assert order.index('parent_end') < order.index('sibling')
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
const ParentEvent = BaseEvent.extend('ParentEvent', {})
const ChildEvent = BaseEvent.extend('ChildEvent', {})
const SiblingEvent = BaseEvent.extend('SiblingEvent', {})
const bus = new EventBus('OrderBus', {
event_concurrency: 'bus-serial',
event_handler_concurrency: 'serial',
})
const order: string[] = []
bus.on(ParentEvent, async (event) => {
order.push('parent_start')
// Detached top-level work: no parent link is recorded for the sibling.
bus.emit(SiblingEvent({}))
const child = event.emit(ChildEvent({}))
await child.now()
order.push('parent_end')
})
bus.on(ChildEvent, async () => {
order.push('child')
})
bus.on(SiblingEvent, async () => {
order.push('sibling')
})
await bus.emit(ParentEvent({})).now()
await bus.waitUntilIdle()
if (!(order.indexOf('child') < order.indexOf('parent_end'))) throw new Error('child should finish before parent resumes')
if (!(order.indexOf('parent_end') < order.indexOf('sibling'))) throw new Error('sibling should run after parent ends')
```
```go theme={null}
bus := abxbus.NewEventBus("OrderBus", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyBusSerial,
EventHandlerConcurrency: abxbus.EventHandlerConcurrencySerial,
})
order := []string{}
bus.On("ParentEvent", "on_parent", func(event *abxbus.BaseEvent) (any, error) {
order = append(order, "parent_start")
// Detached top-level work: no parent link is recorded for the sibling.
bus.Emit(abxbus.NewBaseEvent("SiblingEvent", nil))
child := event.Emit(abxbus.NewBaseEvent("ChildEvent", nil))
if _, err := child.Now(); err != nil {
return nil, err
}
order = append(order, "parent_end")
return nil, nil
}, nil)
bus.On("ChildEvent", "on_child", func(event *abxbus.BaseEvent) (any, error) {
order = append(order, "child")
return nil, nil
}, nil)
bus.On("SiblingEvent", "on_sibling", func(event *abxbus.BaseEvent) (any, error) {
order = append(order, "sibling")
return nil, nil
}, nil)
parent := bus.Emit(abxbus.NewBaseEvent("ParentEvent", nil))
if _, err := parent.Now(); err != nil {
panic(err)
}
if ok := bus.WaitUntilIdle(nil); !ok {
panic("bus did not become idle")
}
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
types::{EventConcurrencyMode, EventHandlerConcurrencyMode},
};
use futures::executor::block_on;
use std::sync::{Arc, Mutex};
event! { struct ParentEvent { event_result_type: () } }
event! { struct ChildEvent { event_result_type: () } }
event! { struct SiblingEvent { event_result_type: () } }
let bus = EventBus::new_with_options(
Some("OrderBus".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::BusSerial,
event_handler_concurrency: EventHandlerConcurrencyMode::Serial,
..EventBusOptions::default()
},
);
let order = Arc::new(Mutex::new(Vec::::new()));
let order_for_parent = order.clone();
let bus_for_parent = bus.clone();
bus.on(ParentEvent, move |event: ParentEvent| {
let order = order_for_parent.clone();
let bus = bus_for_parent.clone();
async move {
order.lock().unwrap().push("parent_start".to_string());
// Detached top-level work: no parent link is recorded for the sibling.
bus.emit(SiblingEvent { ..Default::default() });
let child = event.emit(ChildEvent { ..Default::default() });
child.now().await?;
order.lock().unwrap().push("parent_end".to_string());
Ok(())
}
});
let order_for_child = order.clone();
bus.on(ChildEvent, move |_event: ChildEvent| {
let order = order_for_child.clone();
async move {
order.lock().unwrap().push("child".to_string());
Ok(())
}
});
let order_for_sibling = order.clone();
bus.on(SiblingEvent, move |_event: SiblingEvent| {
let order = order_for_sibling.clone();
async move {
order.lock().unwrap().push("sibling".to_string());
Ok(())
}
});
block_on(bus.emit(ParentEvent { ..Default::default() }).now())?;
assert!(block_on(bus.wait_until_idle(None)));
let order = order.lock().unwrap();
assert!(order.iter().position(|item| item == "child").unwrap() < order.iter().position(|item| item == "parent_end").unwrap());
assert!(order.iter().position(|item| item == "parent_end").unwrap() < order.iter().position(|item| item == "sibling").unwrap());
```
## Interaction with concurrency modes
* `event_concurrency = global-serial`: queue-jump still works, but all buses still share one global event slot.
* `event_concurrency = bus-serial`: queue-jump preempts that bus queue; other buses can continue processing independently.
* `event_concurrency = parallel`: events may already overlap; queue-jump still reduces parent latency for awaited child calls.
* `event_handler_concurrency = serial`: parent temporarily yields execution so child handlers can run without deadlock.
* `event_handler_concurrency = parallel`: child handlers can overlap with other handlers for the same event.
* `event_handler_completion = first`: winner semantics can cancel loser handlers and their in-flight child work.
## Notes
* In Python, `await child_event` inside a handler is the immediate path.
* In Python, `await child_event.wait()` keeps normal queue order (non-queue-jump wait).
* In TypeScript, use `await child_event.now()`.
* In TypeScript, `await child_event.wait()` keeps normal queue order (non-queue-jump wait).
* In Go, use `childEvent.Now()` for the immediate path.
* In Go, use `childEvent.Wait()` to keep normal queue order.
* In Rust, use `child_event.now().await` for the immediate path.
* In Rust, use `child_event.wait().await` to keep normal queue order.
## Related pages
* [Events: global-serial](./events-global-serial)
* [Events: bus-serial](./events-bus-serial)
* [Events: parallel](./events-parallel)
* [Handlers: serial](./handlers-serial)
* [Handlers: parallel](./handlers-parallel)
* [Handler Completion: all](./handler-completion-all)
* [Handler Completion: first](./handler-completion-first)
* [BaseEvent](../api/baseevent)
# Timeout Enforcement
Source: https://abxbus.archivebox.io/concurrency/timeouts
Configure execution deadlines and slow-warning thresholds at bus, event, and handler levels.
Timeout controls operate at three levels:
* Bus defaults (resolved on each bus at processing time when event-level values are unset)
* Per-event overrides (applies to one emitted event instance)
* Per-handler overrides (applies to one handler registration)
Repository example files:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
* [`abxbus-rust/tests/test_eventbus_timeout.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_eventbus_timeout.rs)
* [`examples/log_tree_demo.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/log_tree_demo.py)
* [`abxbus-ts/examples/log_tree_demo.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/log_tree_demo.ts)
## Timeout types
### 1) Event timeout (`event_timeout`)
The outer execution budget for an event. This also acts as an upper cap for each handler run for that event.
### 2) Handler timeout (`event_handler_timeout` / `handler_timeout`)
A handler-specific timeout budget. The effective handler timeout is resolved from handler -> event -> bus, then capped by `event_timeout` when both are set.
### 3) Slow-warning thresholds (`event_slow_timeout`, `event_handler_slow_timeout`, `handler_slow_timeout`)
These emit warnings when work is taking longer than expected:
* `event_slow_timeout`: warns when event processing is still running past the threshold.
* `event_handler_slow_timeout` / `handler_slow_timeout`: warns when a handler run is still running past the threshold.
Slow thresholds are warnings, not forced cancellation.
For event and handler timeout fields, unset (`None` / `null` / `nil`) means inherit the processing bus default at execution time. Set `0` to explicitly disable that timeout or slow warning.
## Where to set each value
| Level | Execution timeout fields | Slow-warning fields |
| - | - | - |
| Bus | `event_timeout` | `event_slow_timeout`, `event_handler_slow_timeout` |
| Event | `event_timeout`, `event_handler_timeout` | `event_slow_timeout`, `event_handler_slow_timeout` |
| Handler | `handler_timeout` | `handler_slow_timeout` |
## Bus-level defaults
Set default budgets and warning thresholds once when creating a bus.
```python theme={null}
from abxbus import EventBus
bus = EventBus(
'TimeoutBus',
event_timeout=30.0,
event_slow_timeout=10.0,
event_handler_slow_timeout=3.0,
)
```
```ts theme={null}
import { EventBus } from 'abxbus'
const bus = new EventBus('TimeoutBus', {
event_timeout: 30,
event_slow_timeout: 10,
event_handler_slow_timeout: 3,
})
```
```go theme={null}
eventTimeout := 30.0
eventSlowTimeout := 10.0
handlerSlowTimeout := 3.0
bus := abxbus.NewEventBus("TimeoutBus", &abxbus.EventBusOptions{
EventTimeout: &eventTimeout,
EventSlowTimeout: &eventSlowTimeout,
EventHandlerSlowTimeout: &handlerSlowTimeout,
})
```
```rust theme={null}
use abxbus::event_bus::{EventBus, EventBusOptions};
let bus = EventBus::new_with_options(
Some("TimeoutBus".to_string()),
EventBusOptions {
event_timeout: Some(30.0),
event_slow_timeout: Some(10.0),
event_handler_slow_timeout: Some(3.0),
..EventBusOptions::default()
},
);
```
## Event-level overrides
Set per-event values when emitting an event instance.
```python theme={null}
from abxbus import BaseEvent
class WorkEvent(BaseEvent):
pass
event = bus.emit(
WorkEvent(
event_timeout=8.0,
event_handler_timeout=2.0,
event_slow_timeout=4.0,
event_handler_slow_timeout=1.0,
)
)
```
```ts theme={null}
import { BaseEvent } from 'abxbus'
const WorkEvent = BaseEvent.extend('WorkEvent', {})
const event = bus.emit(
WorkEvent({
event_timeout: 8,
event_handler_timeout: 2,
event_slow_timeout: 4,
event_handler_slow_timeout: 1,
})
)
```
```go theme={null}
eventTimeout := 8.0
eventHandlerTimeout := 2.0
eventSlowTimeout := 4.0
eventHandlerSlowTimeout := 1.0
event := abxbus.NewBaseEvent("WorkEvent", nil)
event.EventTimeout = &eventTimeout
event.EventHandlerTimeout = &eventHandlerTimeout
event.EventSlowTimeout = &eventSlowTimeout
event.EventHandlerSlowTimeout = &eventHandlerSlowTimeout
event = bus.Emit(event)
```
```rust theme={null}
let event = WorkEvent {
event_timeout: Some(8.0),
event_handler_timeout: Some(2.0),
event_slow_timeout: Some(4.0),
event_handler_slow_timeout: Some(1.0),
..Default::default()
};
let event = bus.emit(event);
```
## Handler-level overrides
Set per-handler timeout and slow-warning overrides at registration time (or by updating the returned handler metadata).
```python theme={null}
entry = bus.on(WorkEvent, slow_handler)
entry.handler_timeout = 1.5
entry.handler_slow_timeout = 0.5
```
```ts theme={null}
bus.on(WorkEvent, slowHandler, {
handler_timeout: 1.5,
handler_slow_timeout: 0.5,
})
```
```go theme={null}
entry := bus.On("WorkEvent", "slow_handler", slowHandler, nil)
handlerTimeout := 1.5
handlerSlowTimeout := 0.5
entry.HandlerTimeout = &handlerTimeout
entry.HandlerSlowTimeout = &handlerSlowTimeout
```
```rust theme={null}
use abxbus::event_handler::EventHandlerOptions;
let entry = bus.on_with_options(
WorkEvent,
"slow_handler",
EventHandlerOptions {
handler_timeout: Some(1.5),
handler_slow_timeout: Some(0.5),
..EventHandlerOptions::default()
},
|_event: WorkEvent| async move { Ok(()) },
);
```
## Precedence rules
### Effective handler timeout
1. Resolve handler timeout source:
* `handler_timeout` (handler level)
* else `event_handler_timeout` (event level)
* else bus `event_timeout`
2. Apply event cap:
* effective timeout is `min(resolved_handler_timeout, event_timeout)` when both are set
* if one is unset, the other value is used
* if both are unset, no timeout is enforced
Resolution happens at processing time on each bus.
For forwarded events, an unset timeout/concurrency field uses the target bus defaults.
### Effective handler slow-warning threshold
Resolved in this order:
1. `handler_slow_timeout`
2. `event_handler_slow_timeout`
3. bus `event_handler_slow_timeout`
### Effective event slow-warning threshold
Resolved in this order:
1. `event_slow_timeout`
2. bus `event_slow_timeout`
## Execution behavior
Timeout and slow-warning behavior is part of the public event lifecycle:
* Event-level timeouts bound the whole event.
* Handler-level timeouts bound each handler and are capped by the remaining event-level budget when both are set.
* Slow-warning settings only log warnings; they do not cancel work.
* Hard timeouts mark unfinished handler results as cancelled, aborted, or timed out and prevent late handler results or late child emits from being accepted.
Event-level timeout finalization keeps cancellation semantics explicit:
* pending handlers -> cancelled
* started handlers -> aborted
## Note on retry
Bus/event timeouts are outer budgets. If you need per-attempt limits for retried handlers, use the `retry` decorator's `timeout` option.
Go does not include the retry decorator yet; for now, wrap retry behavior around your handler body explicitly and keep the bus/event timeout as the outer budget.
# Async and Sync Handlers
Source: https://abxbus.archivebox.io/features/async-sync-handlers
Mix sync and async handlers across functions and method styles.
Python, TypeScript, and Rust support registering sync and async handlers together. Go handlers use one context-aware function shape; handlers can return immediately, block, or coordinate goroutines, while AbxBus controls event/handler concurrency with bus options.
Supported handler shapes:
* bare functions (sync or async)
* static methods (sync or async)
* class-level methods (Python `@classmethod`; in TypeScript, class-level handlers are `static` methods)
* instance methods (sync or async)
```python theme={null}
import asyncio
from abxbus import EventBus, BaseEvent
class WorkEvent(BaseEvent[str]):
task_id: str
def bare_sync(event: WorkEvent) -> str:
return f'bare-sync:{event.task_id}'
async def bare_async(event: WorkEvent) -> str:
await asyncio.sleep(0.01)
return f'bare-async:{event.task_id}'
class HandlerSet:
def __init__(self, prefix: str) -> None:
self.prefix = prefix
@staticmethod
def static_sync(event: WorkEvent) -> str:
return f'static-sync:{event.task_id}'
@staticmethod
async def static_async(event: WorkEvent) -> str:
await asyncio.sleep(0.01)
return f'static-async:{event.task_id}'
@classmethod
def class_sync(cls, event: WorkEvent) -> str:
return f'{cls.__name__}-class-sync:{event.task_id}'
@classmethod
async def class_async(cls, event: WorkEvent) -> str:
await asyncio.sleep(0.01)
return f'{cls.__name__}-class-async:{event.task_id}'
def instance_sync(self, event: WorkEvent) -> str:
return f'{self.prefix}-instance-sync:{event.task_id}'
async def instance_async(self, event: WorkEvent) -> str:
await asyncio.sleep(0.01)
return f'{self.prefix}-instance-async:{event.task_id}'
bus = EventBus('AppBus')
handlers = HandlerSet(prefix='svc')
bus.on(WorkEvent, bare_sync)
bus.on(WorkEvent, bare_async)
bus.on(WorkEvent, HandlerSet.static_sync)
bus.on(WorkEvent, HandlerSet.static_async)
bus.on(WorkEvent, HandlerSet.class_sync)
bus.on(WorkEvent, HandlerSet.class_async)
bus.on(WorkEvent, handlers.instance_sync)
bus.on(WorkEvent, handlers.instance_async)
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const WorkEvent = BaseEvent.extend('WorkEvent', {
task_id: z.string(),
})
const bus = new EventBus('AppBus')
const bareSync = (event: InstanceType) => `bare-sync:${event.task_id}`
const bareAsync = async (event: InstanceType) => {
await new Promise((resolve) => setTimeout(resolve, 10))
return `bare-async:${event.task_id}`
}
class HandlerSet {
constructor(private prefix: string) {}
static staticSync(event: InstanceType) {
return `static-sync:${event.task_id}`
}
static async staticAsync(event: InstanceType) {
await new Promise((resolve) => setTimeout(resolve, 10))
return `static-async:${event.task_id}`
}
instanceSync(event: InstanceType) {
return `${this.prefix}-instance-sync:${event.task_id}`
}
async instanceAsync(event: InstanceType) {
await new Promise((resolve) => setTimeout(resolve, 10))
return `${this.prefix}-instance-async:${event.task_id}`
}
}
const handlers = new HandlerSet('svc')
bus.on(WorkEvent, bareSync)
bus.on(WorkEvent, bareAsync)
bus.on(WorkEvent, HandlerSet.staticSync)
bus.on(WorkEvent, HandlerSet.staticAsync)
bus.on(WorkEvent, handlers.instanceSync.bind(handlers))
bus.on(WorkEvent, handlers.instanceAsync.bind(handlers))
```
```rust theme={null}
use abxbus::{
event,
event_bus::EventBus,
BaseEvent,
};
use futures_timer::Delay;
use std::time::Duration;
event! {
struct WorkEvent {
task_id: String,
event_result_type: String,
}
}
let bus = EventBus::new(Some("AppBus".to_string()));
bus.on(WorkEvent, |event: WorkEvent| async move {
Ok(format!("bare-sync:{}", event.task_id))
});
bus.on(WorkEvent, |event: WorkEvent| async move {
Delay::new(Duration::from_millis(10)).await;
Ok(format!("bare-async:{}", event.task_id))
});
struct HandlerSet {
prefix: String,
}
let handlers = HandlerSet {
prefix: "svc".to_string(),
};
bus.on(WorkEvent, move |event: WorkEvent| async move {
Ok(format!("{}-instance:{}", handlers.prefix, event.task_id))
});
```
```go theme={null}
package main
import (
"fmt"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func bareHandler(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("bare:%s", event.Payload["task_id"]), nil
}
type HandlerSet struct {
Prefix string
}
func StaticStyleHandler(event *abxbus.BaseEvent) (any, error) {
time.Sleep(10 * time.Millisecond)
return fmt.Sprintf("static:%s", event.Payload["task_id"]), nil
}
func (h *HandlerSet) InstanceHandler(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("%s-instance:%s", h.Prefix, event.Payload["task_id"]), nil
}
func main() {
bus := abxbus.NewEventBus("AppBus", nil)
handlers := &HandlerSet{Prefix: "svc"}
bus.On("WorkEvent", "bare_handler", bareHandler, nil)
bus.On("WorkEvent", "static_style_handler", StaticStyleHandler, nil)
bus.On("WorkEvent", "instance_handler", handlers.InstanceHandler, nil)
}
```
# Context Propagation
Source: https://abxbus.archivebox.io/features/context-propagation
Context variables are passed from emit site to event handlers automatically.
## What are Context Variables?
[Context variables](https://nodejs.org/api/async_context.html#asynchronous-context-tracking) are a feature in some languages and runtimes, unrelated to `abxbus`, that let you set ambient "thread-local variables" that may be needed by many functions without passing them manually.
Context variables are mostly used for repetitive arguments that would clutter function signatures if they had to be passed through every layer, such as request IDs, trace and span IDs, user IDs, database transaction IDs, and connection handles.
They are especially common in FastAPI, Fastify, Express/Nest, and OpenTelemetry.
## How `abxbus` works in each runtime
`abxbus` generally snapshots context variables at emit time and provides the snapshot to each event handler when it executes.
For async or queued events, this gives handlers a consistent view of history that matches event dispatch order. This matters because dispatch order is the only strict order that exists when using any of the parallel `event_concurrency` or `event_handler_concurrency` modes.
Event execution order is only statically guaranteed with:
```ts theme={null}
event_concurrency: 'global-serial' | 'bus-serial'
```
These are the fully predictable modes, similar to a database's `STRICT SERIALIZABLE` mode, but they can be impractically slow when processing more than a few thousand events in a complex system. More parallel modes are typically used for performance.
* Python uses [`ContextVars`](https://docs.python.org/3/library/contextvars.html#contextvars.Context) (`contextvars.ContextVar`).
* TypeScript (Node/Bun) uses [`AsyncLocalStorage`](https://nodejs.org/api/async_context.html#asynchronous-context-tracking).
* Rust uses [`thread_local!`](https://doc.rust-lang.org/std/macro.thread_local.html) ([`dcontext`](https://docs.rs/dcontext/latest/dcontext/) snapshots are captured at emit time and restored for handlers).
* Go uses [`context.Context`](https://pkg.go.dev/context) values captured with `EmitWithContext(...)` and restored during handler execution.
## Why this matters
Without propagation, event-based code would be difficult to integrate gradually into larger existing systems.
With propagation, event handlers can log/trace normally like any other request handler or method an existing backend or framework.
```python theme={null}
from contextvars import ContextVar
from abxbus import EventBus, BaseEvent
request_id: ContextVar[str] = ContextVar('request_id', default='')
class RequestEvent(BaseEvent):
pass
bus = EventBus('AppBus')
async def handler(_: RequestEvent) -> None:
print(request_id.get())
# req-123
bus.on(RequestEvent, handler)
request_id.set('req-123')
await bus.emit(RequestEvent()).now()
```
```ts theme={null}
import { AsyncLocalStorage } from 'node:async_hooks'
import { BaseEvent, EventBus } from 'abxbus'
const requestContext = new AsyncLocalStorage<{ requestId: string }>()
const RequestEvent = BaseEvent.extend('RequestEvent', {})
const bus = new EventBus('AppBus')
bus.on(RequestEvent, () => {
console.log(requestContext.getStore()?.requestId)
})
await requestContext.run({ requestId: 'req-123' }, async () => {
await bus.emit(RequestEvent({})).now()
})
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use dcontext::{enter_named_scope, get_context, set_context, try_initialize, RegistryBuilder};
use futures::executor::block_on;
event! {
struct RequestEvent {
event_result_type: (),
}
}
let mut registry = RegistryBuilder::new();
registry.register::("request_id");
let _ = try_initialize(registry);
let bus = EventBus::new(Some("AppBus".to_string()));
bus.on(RequestEvent, |_event: RequestEvent| async move {
println!("{}", get_context::("request_id"));
// req-123
Ok(())
});
let _scope = enter_named_scope("request");
set_context("request_id", "req-123".to_string());
let event = bus.emit(RequestEvent { ..Default::default() });
block_on(event.now());
```
```go theme={null}
package main
import (
"context"
"fmt"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
type contextKey string
func main() {
requestIDKey := contextKey("request_id")
bus := abxbus.NewEventBus("AppBus", nil)
bus.On("RequestEvent", "handler", func(event *abxbus.BaseEvent, ctx context.Context) (any, error) {
fmt.Println(ctx.Value(requestIDKey))
// req-123
return nil, nil
}, nil)
ctx := context.WithValue(context.Background(), requestIDKey, "req-123")
if _, err := bus.EmitWithContext(ctx, abxbus.NewBaseEvent("RequestEvent", nil)).Now(); err != nil {
panic(err)
}
}
```
## Web server style examples
These patterns are typical in frameworks where each incoming request gets a request-local context object.
```python theme={null}
# FastAPI-style shape (conceptual)
request_id.set(incoming_request.headers.get('x-request-id', 'generated-id'))
await bus.emit(RequestEvent()).now()
# handlers can still read request_id.get()
```
```ts theme={null}
// Fastify-style shape (conceptual)
await requestContext.run({ requestId: req.id }, async () => {
await bus.emit(RequestEvent({})).now()
})
// handlers can still read requestContext.getStore()
```
```rust theme={null}
let _scope = enter_named_scope("request");
set_context("request_id", incoming_request_id.to_string());
let event = bus.emit(RequestEvent { ..Default::default() });
block_on(event.now());
// handlers can still read get_context::("request_id")
```
```go theme={null}
// net/http-style shape (conceptual)
ctx := context.WithValue(req.Context(), requestIDKey, req.Header.Get("x-request-id"))
if _, err := bus.EmitWithContext(ctx, abxbus.NewBaseEvent("RequestEvent", nil)).Now(); err != nil {
panic(err)
}
// handlers can still read ctx.Value(requestIDKey)
```
## Browser runtime note
`AsyncLocalStorage` is a Node/Bun API and is not available in browser runtimes.
In browsers:
* AbxBus still works normally for events.
* ambient async context propagation via `AsyncLocalStorage` is not available.
* pass correlation/tracing fields explicitly in event payloads when you need that metadata.
See [Supported Runtimes](../operations/supported-runtimes) for runtime compatibility details.
## Golang runtime note
Go does not have ambient `AsyncLocalStorage`/`ContextVar` state; pass the request context at emit time with `bus.EmitWithContext(...)` or `event.EmitWithContext(...)` and AbxBus propagates that `context.Context` into handlers and awaited child events.
# Event Debouncing
Source: https://abxbus.archivebox.io/features/event-debouncing
Deduplicate expensive event work using find(..., past/future) patterns.
Debouncing is most useful when events trigger expensive work:
* screenshots or browser automation
* external API calls
* LLM/tool runs
* heavyweight DB/file operations
Instead of starting duplicate work every time, reuse:
* a recent matching event (`past` window), or
* a matching event that is about to be emitted by another caller (`future` wait), or
* both (history-first, then short future wait, then emit).
Debouncing in AbxBus is built from `find(...)` + conditional `emit(...)`.
## Debounce building blocks
* `past`: search recent history (`true`/`false`/seconds)
* `future`: optionally wait for a matching future emit (`true`/`false`/seconds)
* `where` / event-field filters: scope matching to the same "work key" (url, account\_id, document\_id, etc.)
See [Find Events](./find-events) for full option semantics.
## Pattern 1: Reuse recent completed work (history-only)
Use when "fresh enough" cached results are acceptable.
```python theme={null}
existing = await bus.find(
ScreenshotEvent,
where=lambda e: e.url == url,
past=10, # look back 10s
future=False, # do not wait
)
event = existing or bus.emit(ScreenshotEvent(url=url))
await event.now()
result = await event.event_result()
```
```ts theme={null}
const existing = await bus.find(
ScreenshotEvent,
(event) => event.url === url,
{ past: 10, future: false }
)
const event = existing ?? bus.emit(ScreenshotEvent({ url }))
await event.now()
const result = await event.eventResult()
```
```rust theme={null}
use abxbus::{
event,
event_bus::FindOptions,
};
use futures::executor::block_on;
use serde_json::json;
use std::collections::HashMap;
event! {
struct ScreenshotEvent {
url: String,
event_result_type: serde_json::Value,
}
}
let existing = block_on(bus.find_with_options(
ScreenshotEvent::event_type,
FindOptions {
past_window: Some(10.0),
future: None,
where_filter: Some(HashMap::from([("url".to_string(), json!(url))])),
..FindOptions::default()
},
));
let event = existing.unwrap_or_else(|| {
bus.emit(ScreenshotEvent { url: url.to_string(), ..Default::default() })
});
block_on(event.now());
let result = block_on(event.event_result())?;
```
```go theme={null}
existing, err := bus.FindEventName("ScreenshotEvent", func(event *abxbus.BaseEvent) bool {
return event.Payload["url"] == url
}, &abxbus.FindOptions{
Past: 10.0, // look back 10s
Future: false, // do not wait
})
if err != nil {
panic(err)
}
event := existing
if event == nil {
event = bus.Emit(abxbus.NewBaseEvent("ScreenshotEvent", map[string]any{"url": url}))
}
if _, err := event.Now(); err != nil {
panic(err)
}
result, err := event.EventResult()
```
## Pattern 2: Coalesce concurrent callers (future-only)
Use when many callers may request the same expensive action at the same time.
Caller A emits first. Caller B waits briefly for that same event instead of emitting a duplicate.
```python theme={null}
in_flight = await bus.find(
ScreenshotEvent,
where=lambda e: e.url == url,
past=False, # skip history
future=2, # wait up to 2s for another caller to emit
)
event = in_flight or bus.emit(ScreenshotEvent(url=url))
await event.now()
result = await event.event_result()
```
```ts theme={null}
const inFlight = await bus.find(
ScreenshotEvent,
(event) => event.url === url,
{ past: false, future: 2 }
)
const event = inFlight ?? bus.emit(ScreenshotEvent({ url }))
await event.now()
const result = await event.eventResult()
```
```rust theme={null}
let in_flight = block_on(bus.find_with_options(
ScreenshotEvent::event_type,
FindOptions {
past: false,
future: Some(2.0),
where_filter: Some(HashMap::from([("url".to_string(), json!(url))])),
..FindOptions::default()
},
));
let event = in_flight.unwrap_or_else(|| {
bus.emit(ScreenshotEvent { url: url.to_string(), ..Default::default() })
});
block_on(event.now());
let result = block_on(event.event_result())?;
```
```go theme={null}
inFlight, err := bus.FindEventName("ScreenshotEvent", func(event *abxbus.BaseEvent) bool {
return event.Payload["url"] == url
}, &abxbus.FindOptions{
Past: false, // skip history
Future: 2.0, // wait up to 2s for another caller to emit
})
if err != nil {
panic(err)
}
event := inFlight
if event == nil {
event = bus.Emit(abxbus.NewBaseEvent("ScreenshotEvent", map[string]any{"url": url}))
}
if _, err := event.Now(); err != nil {
panic(err)
}
result, err := event.EventResult()
```
## Pattern 3: Hybrid debounce (past + short future + emit)
This is the most practical default for expensive endpoints.
1. Reuse recent match.
2. If none, wait briefly for someone else to emit.
3. If still none, emit new work.
```python theme={null}
event = (
await bus.find(ScreenshotEvent, where=lambda e: e.url == url, past=10, future=False)
or await bus.find(ScreenshotEvent, where=lambda e: e.url == url, past=False, future=2)
or bus.emit(ScreenshotEvent(url=url))
)
await event.now()
result = await event.event_result()
```
```ts theme={null}
const event =
(await bus.find(ScreenshotEvent, (e) => e.url === url, { past: 10, future: false })) ??
(await bus.find(ScreenshotEvent, (e) => e.url === url, { past: false, future: 2 })) ??
bus.emit(ScreenshotEvent({ url }))
await event.now()
const result = await event.eventResult()
```
```rust theme={null}
let by_url = HashMap::from([("url".to_string(), json!(url))]);
let event = block_on(bus.find_with_options(
ScreenshotEvent::event_type,
FindOptions {
past_window: Some(10.0),
future: None,
where_filter: Some(by_url.clone()),
..FindOptions::default()
},
))
.or_else(|| {
block_on(bus.find_with_options(
ScreenshotEvent::event_type,
FindOptions {
past: false,
future: Some(2.0),
where_filter: Some(by_url.clone()),
..FindOptions::default()
},
))
})
.unwrap_or_else(|| {
bus.emit(ScreenshotEvent { url: url.to_string(), ..Default::default() })
});
block_on(event.now());
let result = block_on(event.event_result())?;
```
```go theme={null}
sameURL := func(event *abxbus.BaseEvent) bool {
return event.Payload["url"] == url
}
event, err := bus.FindEventName("ScreenshotEvent", sameURL, &abxbus.FindOptions{Past: 10.0, Future: false})
if err != nil {
panic(err)
}
if event == nil {
event, err = bus.FindEventName("ScreenshotEvent", sameURL, &abxbus.FindOptions{Past: false, Future: 2.0})
if err != nil {
panic(err)
}
}
if event == nil {
event = bus.Emit(abxbus.NewBaseEvent("ScreenshotEvent", map[string]any{"url": url}))
}
if _, err := event.Now(); err != nil {
panic(err)
}
result, err := event.EventResult()
```
## Pattern 4: Keyed helper for repeated use
Wrap the debounce logic once and reuse it for all expensive keyed actions.
```python theme={null}
async def emit_debounced_screenshot(url: str):
event = (
await bus.find(ScreenshotEvent, where=lambda e: e.url == url, past=15, future=False)
or await bus.find(ScreenshotEvent, where=lambda e: e.url == url, past=False, future=3)
or bus.emit(ScreenshotEvent(url=url))
)
await event.now()
return await event.event_result()
```
```ts theme={null}
const emitDebouncedScreenshot = async (url: string) => {
const event =
(await bus.find(ScreenshotEvent, (e) => e.url === url, { past: 15, future: false })) ??
(await bus.find(ScreenshotEvent, (e) => e.url === url, { past: false, future: 3 })) ??
bus.emit(ScreenshotEvent({ url }))
await event.now()
return await event.eventResult()
}
```
```rust theme={null}
fn emit_debounced_screenshot(bus: &EventBus, url: &str) -> Result, String> {
let by_url = HashMap::from([("url".to_string(), json!(url))]);
let event = block_on(bus.find_with_options(
ScreenshotEvent::event_type,
FindOptions {
past_window: Some(15.0),
future: None,
where_filter: Some(by_url.clone()),
..FindOptions::default()
},
))
.or_else(|| {
block_on(bus.find_with_options(
ScreenshotEvent::event_type,
FindOptions {
past: false,
future: Some(3.0),
where_filter: Some(by_url),
..FindOptions::default()
},
))
})
.unwrap_or_else(|| {
bus.emit(ScreenshotEvent { url: url.to_string(), ..Default::default() })
});
block_on(event.now());
block_on(event.event_result())
}
```
```go theme={null}
func emitDebouncedScreenshot(ctx context.Context, bus *abxbus.EventBus, url string) (any, error) {
sameURL := func(event *abxbus.BaseEvent) bool {
return event.Payload["url"] == url
}
event, err := bus.FindEventName("ScreenshotEvent", sameURL, &abxbus.FindOptions{Past: 15.0, Future: false})
if err != nil {
return nil, err
}
if event == nil {
event, err = bus.FindEventName("ScreenshotEvent", sameURL, &abxbus.FindOptions{Past: false, Future: 3.0})
if err != nil {
return nil, err
}
}
if event == nil {
event = bus.Emit(abxbus.NewBaseEvent("ScreenshotEvent", map[string]any{"url": url}))
}
if _, err := event.Now(); err != nil {
return nil, err
}
return event.EventResult()
}
```
## Important behavior notes
* `find(...)` resolves when an event is emitted, not when handlers finish.
* Always await completion after selecting a debounced event:
* Python: `await event.now()`, then `await event.event_result()`
* TypeScript: `await event.now()`, then `await event.eventResult()`
* Go: `event.Now()`, then `event.EventResult()`
* Debouncing scope depends on your match key (`where` / event-field filters).\
Use the narrowest key that represents "same work."
* Debouncing depends on retained history. If history is aggressively trimmed, your `past` window can become less effective.
See also:
* [Find Events](./find-events)
* [Event History Store](./event-history-store)
# Event History Store
Source: https://abxbus.archivebox.io/features/event-history-store
Understand queue vs history behavior and how retention settings trim old events.
Each runtime exposes two related (but different) runtime stores:
* `pending_event_queue`: events accepted by the bus but not yet started by the runloop
* `event_history`: events the bus knows about (pending, started, and completed until trimmed)
## What each store is for
| Store | Purpose | Typical contents |
| - | - | - |
| `pending_event_queue` | Scheduling buffer | events waiting their turn to start |
| `event_history` | Observability + lookup | recent pending/started/completed events, bounded by history settings |
The key difference: queue is "what still needs to start", history is "what this bus has seen".
## Retention config options
| Option | Meaning |
| - | - |
| `max_history_size` | Max number of events retained in `event_history` (`null`/`None` means unbounded, `0` means keep only in-flight visibility). |
| `max_history_drop` | If `true`, accept new events and trim oldest history entries when over limit. If `false`, reject new events at the limit (for `max_history_size > 0`). |
## Event lifecycle: queue -> history -> trim
1. Emit:
* Event is accepted.
* Event is added to `event_history`.
* Event is enqueued into `pending_event_queue`.
2. Runloop begins processing:
* Event is removed from `pending_event_queue`.
* Event stays in `event_history` while handlers run.
3. Completion:
* Event is marked completed.
* Event may remain in `event_history` or be dropped based on retention settings.
4. Trimming:
* `max_history_size` and `max_history_drop` determine whether old history is removed or new emits are rejected.
## Trimming behavior by mode
* `max_history_size = None/null`: no automatic history limit.
* `max_history_size = 0`: completed events are removed immediately; only pending/in-flight visibility remains.
* `max_history_size > 0` and `max_history_drop = false`: bus rejects new emits once history reaches the limit.
* `max_history_size > 0` and `max_history_drop = true`: bus trims oldest history entries (prefers completed first; can drop uncompleted entries under extreme pressure).
All runtimes follow this policy. Internally, trim timing is implementation-specific (eager vs amortized cleanup), but externally the semantics above are the contract to rely on.
## Common configurations
```python theme={null}
from abxbus import EventBus
bounded_drop = EventBus(max_history_size=100, max_history_drop=True)
bounded_reject = EventBus(max_history_size=100, max_history_drop=False)
unbounded = EventBus(max_history_size=None)
in_flight_only = EventBus(max_history_size=0)
```
```ts theme={null}
import { EventBus } from 'abxbus'
const boundedDrop = new EventBus('BoundedDropBus', { max_history_size: 100, max_history_drop: true })
const boundedReject = new EventBus('BoundedRejectBus', { max_history_size: 100, max_history_drop: false })
const unbounded = new EventBus('UnboundedBus', { max_history_size: null })
const inFlightOnly = new EventBus('InFlightBus', { max_history_size: 0 })
```
```go theme={null}
maxHistory := 100
zeroHistory := 0
unlimitedHistorySize := abxbus.UnlimitedHistorySize
boundedDrop := abxbus.NewEventBus("BoundedDropBus", &abxbus.EventBusOptions{
MaxHistorySize: &maxHistory,
MaxHistoryDrop: true,
})
boundedReject := abxbus.NewEventBus("BoundedRejectBus", &abxbus.EventBusOptions{
MaxHistorySize: &maxHistory,
MaxHistoryDrop: false,
})
unbounded := abxbus.NewEventBus("UnboundedBus", &abxbus.EventBusOptions{
MaxHistorySize: &unlimitedHistorySize,
})
inFlightOnly := abxbus.NewEventBus("InFlightBus", &abxbus.EventBusOptions{
MaxHistorySize: &zeroHistory,
})
```
```rust theme={null}
use abxbus::event_bus::{EventBus, EventBusOptions};
let bounded_drop = EventBus::new_with_options(Some("BoundedDropBus".to_string()), EventBusOptions {
max_history_size: Some(100),
max_history_drop: true,
..EventBusOptions::default()
});
let bounded_reject = EventBus::new_with_options(Some("BoundedRejectBus".to_string()), EventBusOptions {
max_history_size: Some(100),
max_history_drop: false,
..EventBusOptions::default()
});
let unbounded = EventBus::new_with_options(Some("UnboundedBus".to_string()), EventBusOptions {
max_history_size: None,
..EventBusOptions::default()
});
let in_flight_only = EventBus::new_with_options(Some("InFlightBus".to_string()), EventBusOptions {
max_history_size: Some(0),
..EventBusOptions::default()
});
```
## Inspecting queue vs history at runtime
```python theme={null}
event = bus.emit(MyEvent())
pending_count = bus.pending_event_queue.qsize() if bus.pending_event_queue else 0
history_count = len(bus.event_history)
print('pending_event_queue=', pending_count, 'event_history=', history_count)
# pending_event_queue= 1 event_history= 1
await event.now()
pending_after = bus.pending_event_queue.qsize() if bus.pending_event_queue else 0
history_after = len(bus.event_history)
print('after completion -> pending_event_queue=', pending_after, 'event_history=', history_after)
# after completion -> pending_event_queue= 0 event_history= 1
```
```ts theme={null}
const event = bus.emit(MyEvent({}))
console.log('pending_event_queue=', bus.pending_event_queue.length, 'event_history=', bus.event_history.size)
await event.now()
console.log('after completion -> pending_event_queue=', bus.pending_event_queue.length, 'event_history=', bus.event_history.size)
```
```go theme={null}
event := bus.Emit(abxbus.NewBaseEvent("MyEvent", nil))
var before struct {
PendingEventQueue []string `json:"pending_event_queue"`
EventHistory map[string]any `json:"event_history"`
}
snapshot, err := bus.ToJSON()
if err != nil {
panic(err)
}
if err := json.Unmarshal(snapshot, &before); err != nil {
panic(err)
}
fmt.Println("pending_event_queue=", len(before.PendingEventQueue), "event_history=", len(before.EventHistory))
// pending_event_queue= 1 event_history= 1
if _, err := event.Now(); err != nil {
panic(err)
}
var after struct {
PendingEventQueue []string `json:"pending_event_queue"`
EventHistory map[string]any `json:"event_history"`
}
snapshot, err = bus.ToJSON()
if err != nil {
panic(err)
}
if err := json.Unmarshal(snapshot, &after); err != nil {
panic(err)
}
fmt.Println("after completion -> pending_event_queue=", len(after.PendingEventQueue), "event_history=", len(after.EventHistory))
// after completion -> pending_event_queue= 0 event_history= 1
```
```rust theme={null}
let event = bus.emit(MyEvent {
..Default::default()
});
println!("pending_event_queue/event_history snapshot={}", bus.to_json_value());
block_on(event.now());
println!("after completion -> event_history={}", bus.event_history_size());
```
# Event Pattern Matching
Source: https://abxbus.archivebox.io/features/event-pattern-matching
Use classes, strings, or wildcards for both handler registration and event lookup.
Event patterns are shared across both APIs:
* `bus.on(pattern, handler)` for subscriptions
* `bus.find(pattern, ...)` for history/future lookup
Both accept the same pattern forms:
* event class
* string event type name
* `'*'` wildcard (match everything)
Go and Rust use string event type names for low-level registration/lookup, plus typed payload/event APIs (`bus.On(...)` in Go, `EventSpec`/`BaseEvent` in Rust) when you want payload/result validation.
## Supported pattern forms
| Pattern | Matches | Best for |
| - | - | - |
| Event class (`UserActionEvent`) | One concrete event type | Strong typing end-to-end |
| String (`'UserActionEvent'`) | Events by type name | Dynamic routing/config-driven keys; primary Go/Rust pattern form |
| `'*'` | All event types | Global observers, logging, bridges |
## `.on(...)` and `.find(...)` use the same pattern model
Use whichever operation you need, with the same pattern key:
* subscribe: `bus.on(UserActionEvent, handler)`
* find by class: `await bus.find(UserActionEvent)`
* find by string: `await bus.find('UserActionEvent')`
* wildcard subscribe/find: `bus.on('*', ...)`, `await bus.find('*', ...)`
## Examples
```python theme={null}
from typing import Any
from abxbus import BaseEvent, EventBus
class UserActionEvent(BaseEvent[str]):
action: str
bus = EventBus('AppBus')
async def on(event: UserActionEvent) -> str:
# event is strongly typed here
return f'action:{event.action}'
def on_by_name(event: BaseEvent[Any]) -> None:
# string patterns are looser; payload fields are not statically known
print('by-name', event.event_type, getattr(event, 'action', None))
# by-name UserActionEvent click
def on_any(event: BaseEvent[Any]) -> None:
print('wildcard', event.event_type)
# wildcard UserActionEvent
bus.on(UserActionEvent, on)
bus.on('UserActionEvent', on_by_name)
bus.on('*', on_any)
event = await bus.emit(UserActionEvent(action='click')).now()
result = await event.event_result()
typed_match = await bus.find(UserActionEvent) # UserActionEvent | None
named_match = await bus.find('UserActionEvent') # BaseEvent[Any] | None
wildcard_match = await bus.find('*', future=5) # BaseEvent[Any] | None
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const UserActionEvent = BaseEvent.extend('UserActionEvent', {
action: z.string(),
event_result_type: z.string(),
})
const bus = new EventBus('AppBus')
bus.on(UserActionEvent, (event) => {
// event is strongly typed here
return `action:${event.action}`
})
bus.on('UserActionEvent', (event) => {
// string patterns are looser; event is BaseEvent-like at compile time
console.log('by-name', event.event_type)
return undefined
})
bus.on('*', (event) => {
console.log('wildcard', event.event_type)
return undefined
})
const typedMatch = await bus.find(UserActionEvent) // InstanceType | null
const namedMatch = await bus.find('UserActionEvent') // BaseEvent | null
const wildcardMatch = await bus.find('*', { future: 5 }) // BaseEvent | null
```
```rust theme={null}
use abxbus::{
event,
event_bus::EventBus,
BaseEvent,
};
use futures::executor::block_on;
event! {
struct UserActionEvent {
action: String,
event_result_type: String,
}
}
let bus = EventBus::new(Some("AppBus".to_string()));
bus.on(UserActionEvent, |event: UserActionEvent| async move {
Ok(format!("action:{}", event.action))
});
let event = bus.emit(UserActionEvent {
action: "click".to_string(),
..Default::default()
});
block_on(event.now());
let typed_match = block_on(bus.find::(true, None));
let named_match = block_on(bus.find("UserActionEvent", true, None, None));
let wildcard_match = block_on(bus.find("*", false, Some(5.0), None));
```
```go theme={null}
package main
import (
"fmt"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
type UserActionEvent struct {
Action string `json:"action"`
}
func main() {
bus := abxbus.NewEventBus("AppBus", nil)
bus.On(func(event UserActionEvent) (string, error) {
return fmt.Sprintf("action:%s", event.Action), nil
})
bus.On("UserActionEvent", "on_by_name", func(event *abxbus.BaseEvent) (any, error) {
fmt.Println("by-name", event.EventType, event.Payload["action"])
// by-name UserActionEvent click
return nil, nil
}, nil)
bus.OnEventName("*", "on_any", func(event *abxbus.BaseEvent) (any, error) {
fmt.Println("wildcard", event.EventType)
// wildcard UserActionEvent
return nil, nil
}, nil)
_, err := bus.Emit(UserActionEvent{Action: "click"}).EventResult()
if err != nil {
panic(err)
}
namedMatch, err := bus.FindEventName("UserActionEvent", nil, nil)
if err != nil {
panic(err)
}
wildcardMatch, err := bus.FindEventName("*", nil, &abxbus.FindOptions{Future: 5.0})
if err != nil {
panic(err)
}
_, _ = namedMatch, wildcardMatch
}
```
## Why event classes are preferred for typing
Event classes preserve the most useful static typing:
* handler input shape is specific (payload fields are known)
* event result typing stays aligned with `event_result_type` / generic result type
* `.find(EventClass)` returns the specific event type
String keys and `'*'` are intentionally looser:
* Python: treat as `BaseEvent[Any]`
* TypeScript: typed as base `BaseEvent`/unknown-oriented handler return checks
* Rust: typed handlers use `bus.on(MyEvent, ...)`; raw string/wildcard registration is reserved for low-level forwarding internals.
* Go: typed handlers use `bus.On(...)`; string/wildcard handlers use `bus.OnEventName(...)` and receive `*BaseEvent`
Use string/wildcard patterns when you need dynamic behavior. Use classes whenever you want strict payload/result type hints through handlers and lookups.
# FIFO Event Processing
Source: https://abxbus.archivebox.io/features/fifo-processing
Process queued events in deterministic first-in-first-out order.
Using the default options out-of-the-box, all events and handlers on a bus process in strict serial order to make execution order predictable and consistency easy.
This is the default behavior because:
* `event_concurrency='bus-serial'`
* `event_handler_concurrency='serial'`
* `event_handler_completion='all'`
On a single bus, that means event `N+1` never starts before event `N` is complete, even if event `N+1` handlers are "faster".
As you scale, you can tune these guarantees. See [Concurrency Control](../concurrency/immediate-execution) in the sidebar for all modes and tradeoffs.
## Variable handler runtimes still stay FIFO
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class JobEvent(BaseEvent):
order: int
delay_s: float
bus = EventBus('FifoBus')
started_order: list[int] = []
completed_order: list[int] = []
async def on_job(event: JobEvent) -> None:
started_order.append(event.order)
await asyncio.sleep(event.delay_s)
completed_order.append(event.order)
bus.on(JobEvent, on_job)
emitted = [
bus.emit(JobEvent(order=0, delay_s=0.030)),
bus.emit(JobEvent(order=1, delay_s=0.001)),
bus.emit(JobEvent(order=2, delay_s=0.020)),
]
await bus.wait_until_idle()
print(started_order)
# [0, 1, 2]
print(completed_order)
# [0, 1, 2]
print([event.event_started_at is not None for event in emitted])
# [True, True, True]
print([event.event_completed_at is not None for event in emitted])
# [True, True, True]
print(emitted[0].event_started_at <= emitted[1].event_started_at <= emitted[2].event_started_at)
# True
print(emitted[0].event_completed_at <= emitted[1].event_completed_at <= emitted[2].event_completed_at)
# True
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const JobEvent = BaseEvent.extend('JobEvent', {
order: z.number(),
delay_ms: z.number(),
})
const bus = new EventBus('FifoBus')
const startedOrder: number[] = []
const completedOrder: number[] = []
bus.on(JobEvent, async (event) => {
startedOrder.push(event.order)
await new Promise((resolve) => setTimeout(resolve, event.delay_ms))
completedOrder.push(event.order)
})
const emitted = [
bus.emit(JobEvent({ order: 0, delay_ms: 30 })),
bus.emit(JobEvent({ order: 1, delay_ms: 1 })),
bus.emit(JobEvent({ order: 2, delay_ms: 20 })),
]
await bus.waitUntilIdle()
console.log(startedOrder)
// [0, 1, 2]
console.log(completedOrder)
// [0, 1, 2]
console.log(emitted.map((event) => Boolean(event.event_started_at)))
// [true, true, true]
console.log(emitted.map((event) => Boolean(event.event_completed_at)))
// [true, true, true]
console.log(
Date.parse(emitted[0].event_started_at!) <=
Date.parse(emitted[1].event_started_at!) &&
Date.parse(emitted[1].event_started_at!) <= Date.parse(emitted[2].event_started_at!)
)
// true
console.log(
Date.parse(emitted[0].event_completed_at!) <=
Date.parse(emitted[1].event_completed_at!) &&
Date.parse(emitted[1].event_completed_at!) <= Date.parse(emitted[2].event_completed_at!)
)
// true
```
```go theme={null}
package main
import (
"fmt"
"reflect"
"time"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("FifoBus", nil)
startedOrder := []int{}
completedOrder := []int{}
bus.On("JobEvent", "on_job", func(event *abxbus.BaseEvent) (any, error) {
order := event.Payload["order"].(int)
delayMS := event.Payload["delay_ms"].(int)
startedOrder = append(startedOrder, order)
time.Sleep(time.Duration(delayMS) * time.Millisecond)
completedOrder = append(completedOrder, order)
return nil, nil
}, nil)
emitted := []*abxbus.BaseEvent{
bus.Emit(abxbus.NewBaseEvent("JobEvent", map[string]any{"order": 0, "delay_ms": 30})),
bus.Emit(abxbus.NewBaseEvent("JobEvent", map[string]any{"order": 1, "delay_ms": 1})),
bus.Emit(abxbus.NewBaseEvent("JobEvent", map[string]any{"order": 2, "delay_ms": 20})),
}
bus.WaitUntilIdle(nil)
fmt.Println(startedOrder)
// [0 1 2]
fmt.Println(completedOrder)
// [0 1 2]
if !reflect.DeepEqual(startedOrder, []int{0, 1, 2}) {
panic("started order is not FIFO")
}
if !reflect.DeepEqual(completedOrder, []int{0, 1, 2}) {
panic("completed order is not FIFO")
}
fmt.Println(emitted[0].EventStartedAt != nil, emitted[1].EventStartedAt != nil, emitted[2].EventStartedAt != nil)
// true true true
}
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
use std::sync::{Arc, Mutex};
event! {
struct JobEvent {
order: i64,
event_result_type: (),
}
}
let bus = EventBus::new(Some("FifoBus".to_string()));
let started_order = Arc::new(Mutex::new(Vec::new()));
let completed_order = Arc::new(Mutex::new(Vec::new()));
let started = started_order.clone();
let completed = completed_order.clone();
bus.on(JobEvent, move |event: JobEvent| async move {
let order = event.order;
started.lock().unwrap().push(order);
completed.lock().unwrap().push(order);
Ok(())
});
for order in 0..3 {
bus.emit(JobEvent { order, ..Default::default() });
}
block_on(bus.wait_until_idle(None));
assert_eq!(*started_order.lock().unwrap(), vec![0, 1, 2]);
assert_eq!(*completed_order.lock().unwrap(), vec![0, 1, 2]);
```
## Ambiguous case: slow then fast still runs serially
Even if you emit a slow event and then a fast event right after, the fast one does not overtake on the same bus under defaults.
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class SlowEvent(BaseEvent):
name: str
class FastEvent(BaseEvent):
name: str
bus = EventBus('FifoBus')
trace: list[str] = []
async def on_slow(event: SlowEvent) -> None:
trace.append(f'start:{event.event_type}:{event.name}')
await asyncio.sleep(0.040)
trace.append(f'end:{event.event_type}:{event.name}')
async def on_fast(event: FastEvent) -> None:
trace.append(f'start:{event.event_type}:{event.name}')
await asyncio.sleep(0.001)
trace.append(f'end:{event.event_type}:{event.name}')
bus.on(SlowEvent, on_slow)
bus.on(FastEvent, on_fast)
slow = bus.emit(SlowEvent(name='slow-a'))
fast = bus.emit(FastEvent(name='fast-b'))
await bus.wait_until_idle()
print(trace)
# ['start:SlowEvent:slow-a', 'end:SlowEvent:slow-a', 'start:FastEvent:fast-b', 'end:FastEvent:fast-b']
print(slow.event_completed_at <= fast.event_started_at)
# True
tree_lines = [
line for line in bus.log_tree().splitlines()
if 'SlowEvent#' in line or 'FastEvent#' in line
]
print(tree_lines)
# ['├── SlowEvent#6aa1 [14:09:10.120 (0.040s)]', '└── FastEvent#6aa2 [14:09:10.161 (0.001s)]']
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const SlowEvent = BaseEvent.extend('SlowEvent', {
name: z.string(),
})
const FastEvent = BaseEvent.extend('FastEvent', {
name: z.string(),
})
const bus = new EventBus('FifoBus')
const trace: string[] = []
bus.on(SlowEvent, async (event) => {
trace.push(`start:${event.event_type}:${event.name}`)
await new Promise((resolve) => setTimeout(resolve, 40))
trace.push(`end:${event.event_type}:${event.name}`)
})
bus.on(FastEvent, async (event) => {
trace.push(`start:${event.event_type}:${event.name}`)
await new Promise((resolve) => setTimeout(resolve, 1))
trace.push(`end:${event.event_type}:${event.name}`)
})
const slow = bus.emit(SlowEvent({ name: 'slow-a' }))
const fast = bus.emit(FastEvent({ name: 'fast-b' }))
await bus.waitUntilIdle()
console.log(trace)
// ['start:SlowEvent:slow-a', 'end:SlowEvent:slow-a', 'start:FastEvent:fast-b', 'end:FastEvent:fast-b']
console.log(Date.parse(slow.event_completed_at!) <= Date.parse(fast.event_started_at!))
// true
const treeLines = bus
.logTree()
.split('\n')
.filter((line) => line.includes('SlowEvent#') || line.includes('FastEvent#'))
console.log(treeLines)
// ['├── ✅ SlowEvent#6aa1 [14:09:10.120 (0.040s)]', '└── ✅ FastEvent#6aa2 [14:09:10.161 (0.001s)]']
```
```go theme={null}
bus := abxbus.NewEventBus("FifoBus", nil)
trace := []string{}
bus.On("SlowEvent", "on_slow", func(event *abxbus.BaseEvent) (any, error) {
trace = append(trace, fmt.Sprintf("start:%s:%s", event.EventType, event.Payload["name"]))
time.Sleep(40 * time.Millisecond)
trace = append(trace, fmt.Sprintf("end:%s:%s", event.EventType, event.Payload["name"]))
return nil, nil
}, nil)
bus.On("FastEvent", "on_fast", func(event *abxbus.BaseEvent) (any, error) {
trace = append(trace, fmt.Sprintf("start:%s:%s", event.EventType, event.Payload["name"]))
time.Sleep(1 * time.Millisecond)
trace = append(trace, fmt.Sprintf("end:%s:%s", event.EventType, event.Payload["name"]))
return nil, nil
}, nil)
slow := bus.Emit(abxbus.NewBaseEvent("SlowEvent", map[string]any{"name": "slow-a"}))
fast := bus.Emit(abxbus.NewBaseEvent("FastEvent", map[string]any{"name": "fast-b"}))
bus.WaitUntilIdle(nil)
fmt.Println(trace)
// [start:SlowEvent:slow-a end:SlowEvent:slow-a start:FastEvent:fast-b end:FastEvent:fast-b]
fmt.Println(*slow.EventCompletedAt <= *fast.EventStartedAt)
// true
fmt.Println(bus.LogTree())
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
use std::sync::{Arc, Mutex};
event! {
struct SlowEvent {
event_result_type: (),
}
}
event! {
struct FastEvent {
event_result_type: (),
}
}
let bus = EventBus::new(Some("FifoBus".to_string()));
let trace = Arc::new(Mutex::new(Vec::new()));
let trace_for_slow = trace.clone();
bus.on(SlowEvent, move |_event: SlowEvent| async move {
trace_for_slow.lock().unwrap().push("start:SlowEvent".to_string());
trace_for_slow.lock().unwrap().push("end:SlowEvent".to_string());
Ok(())
});
let trace_for_fast = trace.clone();
bus.on(FastEvent, move |_event: FastEvent| async move {
trace_for_fast.lock().unwrap().push("start:FastEvent".to_string());
trace_for_fast.lock().unwrap().push("end:FastEvent".to_string());
Ok(())
});
let slow = bus.emit(SlowEvent { ..Default::default() });
let fast = bus.emit(FastEvent { ..Default::default() });
block_on(bus.wait_until_idle(None));
assert!(slow.event_completed_at.read() <= fast.event_started_at.read());
println!("{}", bus.log_tree());
```
## Important exception: awaited child events
Inside a running handler, if you emit and await a child event, that child can queue-jump for RPC-style behavior. This is the intentional exception to plain FIFO queue order.
See [Immediate Execution (RPC-style)](../concurrency/immediate-execution) for exact behavior and mode interactions.
# Find Events
Source: https://abxbus.archivebox.io/features/find-events
Query history and optionally wait for matching future events.
`find(...)` is the unified lookup API: search history, wait for future events, or combine both.
## Interface
```python theme={null}
await bus.find(
event_type, # Event class, event type string, or '*'
where: Callable[[BaseEvent], bool] | None = None,
child_of: BaseEvent | None = None,
past: bool | float | timedelta = True,
future: bool | float = False,
**event_fields, # equality filters (event_status='completed', request_id='abc', ...)
)
```
```ts theme={null}
await bus.find(event_pattern, options?)
await bus.find(event_pattern, where, options?)
// options:
{
past?: boolean | number // seconds when number
future?: boolean | number // seconds when number
child_of?: BaseEvent | null
[event_field: string]: unknown // equality filters, e.g. event_status: 'completed'
}
```
```go theme={null}
bus.FindEventName(
"ResponseEvent", // event type string or "*"
func(event *abxbus.BaseEvent) bool { // optional predicate
return true
},
&abxbus.FindOptions{
Past: true, // bool or seconds as float64/int
Future: false, // bool or seconds as float64/int
ChildOf: parentEvent, // optional lineage filter
Equals: map[string]any{ // event metadata or payload equality filters
"event_status": "completed",
"request_id": "abc",
},
},
)
```
```rust theme={null}
use abxbus::event_bus::FindOptions;
use futures::executor::block_on;
use serde_json::json;
use std::{collections::HashMap, sync::Arc};
block_on(bus.find(
"ResponseEvent", // event type string or "*"
true, // past history lookup
Some(5.0), // future wait timeout in seconds
None, // child_of parent event
));
block_on(bus.find_with_options(
"ResponseEvent",
FindOptions {
past: false,
future: Some(5.0),
child_of: Some(parent_event.clone()),
where_filter: Some(HashMap::from([
("event_status".to_string(), json!("completed")),
("request_id".to_string(), json!("abc")),
])),
..FindOptions::default()
},
));
```
## Option semantics
* `past`
* `true`: search all history (default)
* `false`: skip history
* `number` (or `timedelta` in Python): search recent history window
* `future`
* `false`: do not wait (default)
* `true`: wait indefinitely
* `number`: wait up to N seconds
* `where`: predicate filter
* `child_of`: match only descendants of the given parent event
* `event_fields`: strict equality filters on event fields/metadata
Default behavior when omitted is history-only lookup (`past=True`, `future=False`).
## Common use cases
### 1) History lookup only (non-blocking)
```python theme={null}
existing = await bus.find(ResponseEvent)
```
```ts theme={null}
const existing = await bus.find(ResponseEvent)
```
```go theme={null}
existing, err := bus.FindEventName("ResponseEvent", nil, nil)
```
```rust theme={null}
let existing = block_on(bus.find("ResponseEvent", true, None, None));
```
### 2) Wait only for future events
```python theme={null}
future = await bus.find(ResponseEvent, past=False, future=5)
```
```ts theme={null}
const future = await bus.find(ResponseEvent, { past: false, future: 5 })
```
```go theme={null}
future, err := bus.FindEventName("ResponseEvent", nil, &abxbus.FindOptions{
Past: false,
Future: 5.0,
})
```
```rust theme={null}
let future = block_on(bus.find("ResponseEvent", false, Some(5.0), None));
```
### 3) Check recent history, then keep waiting briefly
```python theme={null}
match = await bus.find(ResponseEvent, past=5, future=5)
```
```ts theme={null}
const match = await bus.find(ResponseEvent, { past: 5, future: 5 })
```
```go theme={null}
match, err := bus.FindEventName("ResponseEvent", nil, &abxbus.FindOptions{
Past: 5.0,
Future: 5.0,
})
```
```rust theme={null}
let match_event = block_on(bus.find_with_options(
"ResponseEvent",
FindOptions {
past_window: Some(5.0),
future: Some(5.0),
..FindOptions::default()
},
));
```
### 4) Filter by fields + predicate
```python theme={null}
match = await bus.find(
ResponseEvent,
where=lambda e: e.request_id == my_id,
event_status='completed',
future=5,
)
```
```ts theme={null}
const match = await bus.find(
ResponseEvent,
(event) => event.request_id === myId,
{ event_status: 'completed', future: 5 }
)
```
```go theme={null}
match, err := bus.FindEventName(
"ResponseEvent",
func(event *abxbus.BaseEvent) bool {
return event.Payload["request_id"] == myID
},
&abxbus.FindOptions{
Future: 5.0,
Equals: map[string]any{"event_status": "completed"},
},
)
```
```rust theme={null}
let match_event = block_on(bus.find_with_options(
"ResponseEvent",
FindOptions {
future: Some(5.0),
where_filter: Some(HashMap::from([(
"request_id".to_string(),
json!(my_id),
)])),
..FindOptions::default()
},
));
```
### 5) Wildcard lookup across all event types
```python theme={null}
any_completed = await bus.find(
'*',
where=lambda e: e.event_type.endswith('ResultEvent'),
event_status='completed',
future=5,
)
```
```ts theme={null}
const anyCompleted = await bus.find(
'*',
(event) => event.event_type.endsWith('ResultEvent'),
{ event_status: 'completed', future: 5 }
)
```
```go theme={null}
anyCompleted, err := bus.FindEventName(
"*",
func(event *abxbus.BaseEvent) bool {
return strings.HasSuffix(event.EventType, "ResultEvent")
},
&abxbus.FindOptions{
Future: 5.0,
Equals: map[string]any{"event_status": "completed"},
},
)
```
```rust theme={null}
let any_completed = block_on(bus.find_with_options(
"*",
FindOptions {
future: Some(5.0),
where_filter: Some(HashMap::from([(
"event_status".to_string(),
json!("completed"),
)])),
where_predicate: Some(Arc::new(|event| {
event
.to_json_value()
.get("event_type")
.and_then(|value| value.as_str())
.is_some_and(|event_type| event_type.ends_with("ResultEvent"))
})),
..FindOptions::default()
},
));
```
### 6) Find descendants of a specific parent event
```python theme={null}
parent_event = await bus.emit(NavigateToUrlEvent(url='https://example.com')).now()
child = await bus.find(TabCreatedEvent, child_of=parent_event, past=5)
```
```ts theme={null}
const parentEvent = await bus.emit(NavigateToUrlEvent({ url: 'https://example.com' })).now()
const child = await bus.find(TabCreatedEvent, { child_of: parentEvent, past: 5 })
```
```go theme={null}
parentEvent := bus.Emit(abxbus.NewBaseEvent("NavigateToUrlEvent", map[string]any{
"url": "https://example.com",
}))
if _, err := parentEvent.Now(); err != nil {
panic(err)
}
child, err := bus.FindEventName("TabCreatedEvent", nil, &abxbus.FindOptions{
ChildOf: parentEvent,
Past: 5.0,
})
```
```rust theme={null}
event! {
struct NavigateToUrlEvent {
url: String,
event_result_type: (),
}
}
event! {
struct TabCreatedEvent {
event_result_type: (),
}
}
let parent_event = bus.emit(NavigateToUrlEvent {
url: "https://example.com".to_string(),
..Default::default()
});
block_on(parent_event.now())?;
let parent_ref = block_on(bus.find("NavigateToUrlEvent", true, None, None))
.expect("parent event should be in history");
let child = block_on(bus.find_with_options(
"TabCreatedEvent",
FindOptions {
past_window: Some(5.0),
child_of: Some(parent_ref),
..FindOptions::default()
},
));
```
### 7) Debounce expensive work
```python theme={null}
event = (
await bus.find(ScreenshotEvent, past=10, future=False)
or await bus.find(ScreenshotEvent, past=False, future=5)
or bus.emit(ScreenshotEvent())
)
await event.now()
```
```ts theme={null}
const event =
(await bus.find(ScreenshotEvent, { past: 10, future: false })) ??
(await bus.find(ScreenshotEvent, { past: false, future: 5 })) ??
bus.emit(ScreenshotEvent({}))
await event.now()
```
```go theme={null}
event, err := bus.FindEventName("ScreenshotEvent", nil, &abxbus.FindOptions{Past: 10.0, Future: false})
if err != nil {
panic(err)
}
if event == nil {
event, err = bus.FindEventName("ScreenshotEvent", nil, &abxbus.FindOptions{Past: false, Future: 5.0})
if err != nil {
panic(err)
}
}
if event == nil {
event = bus.Emit(abxbus.NewBaseEvent("ScreenshotEvent", nil))
}
if _, err := event.Now(); err != nil {
panic(err)
}
```
```rust theme={null}
event! {
struct ScreenshotEvent {
event_result_type: Vec,
}
}
let event = match block_on(bus.find_with_options(
"ScreenshotEvent",
FindOptions {
past_window: Some(10.0),
future: None,
..FindOptions::default()
},
)) {
Some(raw) => serde_json::from_value::(raw.to_json_value())?,
None => match block_on(bus.find_with_options(
"ScreenshotEvent",
FindOptions {
past: false,
future: Some(5.0),
..FindOptions::default()
},
)) {
Some(raw) => serde_json::from_value::(raw.to_json_value())?,
None => bus.emit(ScreenshotEvent { ..Default::default() }),
},
};
block_on(event.now())?;
```
## Important behavior
* `find()` resolves when an event is emitted, not when handlers finish.
* To wait for handler completion, call `await event.now()` in Python/TypeScript, `event.Now()` in Go, or `event.now().await` in Rust.
* If no match is found (or `future` times out), `find()` returns `None` / `null`.
* If both `past` and `future` are `false`, it returns immediately with no match.
## Returning multiple matches: `filter()`
`filter()` takes the same arguments as `find()` but returns the list of all matching events
(ordered newest to oldest), plus an optional `limit` argument to cap the number of results.
`find()` is equivalent to `filter(..., limit=1)` returning the first match (or `None` / `null`).
```python theme={null}
recent = await bus.filter(ResponseEvent, past=10, future=False, limit=5)
```
```ts theme={null}
const recent = await bus.filter(ResponseEvent, { past: 10, future: false, limit: 5 })
```
```go theme={null}
limit := 5
recent, err := bus.FilterEventName("ResponseEvent", nil, &abxbus.FilterOptions{
Past: 10.0,
Future: false,
Limit: &limit,
})
```
```rust theme={null}
use abxbus::event_bus::FilterOptions;
let recent = block_on(bus.filter_with_options(
"ResponseEvent",
FilterOptions {
past_window: Some(10.0),
future: None,
limit: Some(5),
..FilterOptions::default()
},
));
```
# Forwarding Between Buses
Source: https://abxbus.archivebox.io/features/forwarding-between-buses
Compose multiple buses with automatic forwarding loop prevention.
You can forward events across multiple buses while preserving event path metadata and loop safety.
Forwarding does not eagerly hydrate unset event options. Fields such as `event_timeout`, `event_concurrency`, `event_handler_concurrency`, and `event_handler_completion` stay unset on the event unless the user set them, and each processing bus applies its own defaults at execution time.
Repository example files:
* [`examples/forwarding_between_busses.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/forwarding_between_busses.py)
* [`abxbus-ts/examples/forwarding_between_busses.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/forwarding_between_busses.ts)
* [`abxbus-rust/tests/test_eventbus_forwarding.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_eventbus_forwarding.rs)
## Why multiple buses are useful
Multiple buses let you separate concerns and tune runtime behavior per boundary:
* service-local bus for business logic with strict ordering and useful history
* transport/relay bus focused on throughput and forwarding (little or no history retention)
* specialized buses for domains that need different timeout or concurrency policies
This is especially useful in microservice-style designs, where each component has different consistency and observability needs.
## Example: service buses with different policies
In this example:
* `AuthBus` is strict and debuggable: `event_concurrency='bus-serial'`, `event_handler_concurrency='serial'`, `max_history_size=100`
* `RelayBus` is a transport forwarder: `event_concurrency='parallel'`, `max_history_size=0`
* `BillingBus` is another service bus with its own settings
```python theme={null}
from abxbus import BaseEvent, EventBus
class UserCreatedEvent(BaseEvent[str]):
user_id: str
class AuthService:
def __init__(self) -> None:
self.bus = EventBus(
'AuthBus',
event_concurrency='bus-serial',
event_handler_concurrency='serial',
max_history_size=100,
)
self.bus.on(UserCreatedEvent, self.on_user_created)
async def on_user_created(self, event: UserCreatedEvent) -> str:
return f'auth-ok:{event.user_id}'
class RelayService:
def __init__(self) -> None:
self.bus = EventBus(
'RelayBus',
event_concurrency='parallel',
max_history_size=0,
)
class BillingService:
def __init__(self) -> None:
self.bus = EventBus(
'BillingBus',
event_concurrency='bus-serial',
event_handler_concurrency='serial',
max_history_size=100,
)
self.bus.on(UserCreatedEvent, self.on_user_created)
async def on_user_created(self, event: UserCreatedEvent) -> str:
return f'billing-ok:{event.user_id}'
auth = AuthService()
relay = RelayService()
billing = BillingService()
auth.bus.on('*', relay.bus.emit)
relay.bus.on('*', billing.bus.emit)
event = await auth.bus.emit(UserCreatedEvent(user_id='u-a8d1')).now(first_result=True)
result = await event.event_result()
print(result)
# 'auth-ok:u-a8d1'
root = auth.bus.emit(UserCreatedEvent(user_id='u-a8d1'))
await root.now()
print(root.event_path)
# ['AuthBus#a8d1', 'RelayBus#3f2c', 'BillingBus#b91e']
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const UserCreatedEvent = BaseEvent.extend('UserCreatedEvent', {
user_id: z.string(),
event_result_type: z.string(),
})
class AuthService {
bus = new EventBus('AuthBus', {
event_concurrency: 'bus-serial',
event_handler_concurrency: 'serial',
max_history_size: 100,
})
constructor() {
this.bus.on(UserCreatedEvent, this.onUserCreated)
}
onUserCreated = async (event: InstanceType) => `auth-ok:${event.user_id}`
}
class RelayService {
bus = new EventBus('RelayBus', {
event_concurrency: 'parallel',
max_history_size: 0,
})
}
class BillingService {
bus = new EventBus('BillingBus', {
event_concurrency: 'bus-serial',
event_handler_concurrency: 'serial',
max_history_size: 100,
})
constructor() {
this.bus.on(UserCreatedEvent, this.onUserCreated)
}
onUserCreated = async (event: InstanceType) => `billing-ok:${event.user_id}`
}
const auth = new AuthService()
const relay = new RelayService()
const billing = new BillingService()
auth.bus.on('*', relay.bus.emit)
relay.bus.on('*', billing.bus.emit)
const event = auth.bus.emit(UserCreatedEvent({ user_id: 'u-a8d1' }))
await event.now()
console.log(await event.eventResult())
// 'auth-ok:u-a8d1'
console.log(event.event_path)
// ['AuthBus#a8d1', 'RelayBus#3f2c', 'BillingBus#b91e']
```
```rust theme={null}
use abxbus::{
event,
event_bus::{EventBus, EventBusOptions},
types::{EventConcurrencyMode, EventHandlerConcurrencyMode},
};
use futures::executor::block_on;
use serde_json::json;
event! {
struct UserCreatedEvent {
user_id: String,
event_result_type: serde_json::Value,
}
}
let auth = EventBus::new_with_options(
Some("AuthBus".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::BusSerial,
event_handler_concurrency: EventHandlerConcurrencyMode::Serial,
max_history_size: Some(100),
..EventBusOptions::default()
},
);
let relay = EventBus::new_with_options(
Some("RelayBus".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::Parallel,
max_history_size: Some(0),
..EventBusOptions::default()
},
);
let billing = EventBus::new_with_options(
Some("BillingBus".to_string()),
EventBusOptions {
event_concurrency: EventConcurrencyMode::BusSerial,
event_handler_concurrency: EventHandlerConcurrencyMode::Serial,
max_history_size: Some(100),
..EventBusOptions::default()
},
);
auth.on(UserCreatedEvent, |event: UserCreatedEvent| async move {
Ok(json!(format!("auth-ok:{}", event.user_id)))
});
billing.on(UserCreatedEvent, |event: UserCreatedEvent| async move {
Ok(json!(format!("billing-ok:{}", event.user_id)))
});
let relay_for_handler = relay.clone();
auth.on(UserCreatedEvent, move |event: UserCreatedEvent| {
let relay_for_handler = relay_for_handler.clone();
async move {
Ok(relay_for_handler.emit(event).to_json_value())
}
});
let billing_for_handler = billing.clone();
relay.on(UserCreatedEvent, move |event: UserCreatedEvent| {
let billing_for_handler = billing_for_handler.clone();
async move {
Ok(billing_for_handler.emit(event).to_json_value())
}
});
let event = auth.emit(UserCreatedEvent {
user_id: "u-a8d1".to_string(),
..Default::default()
});
let result = block_on(event.event_result())?;
block_on(event.now());
println!("{result}");
println!("{:?}", event.event_path.read());
```
```go theme={null}
package main
import (
"fmt"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
type AuthService struct {
bus *abxbus.EventBus
}
type RelayService struct {
bus *abxbus.EventBus
}
type BillingService struct {
bus *abxbus.EventBus
}
func main() {
maxHistory := 100
zeroHistory := 0
auth := &AuthService{bus: abxbus.NewEventBus("AuthBus", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyBusSerial,
EventHandlerConcurrency: abxbus.EventHandlerConcurrencySerial,
MaxHistorySize: &maxHistory,
})}
relay := &RelayService{bus: abxbus.NewEventBus("RelayBus", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyParallel,
MaxHistorySize: &zeroHistory,
})}
billing := &BillingService{bus: abxbus.NewEventBus("BillingBus", &abxbus.EventBusOptions{
EventConcurrency: abxbus.EventConcurrencyBusSerial,
EventHandlerConcurrency: abxbus.EventHandlerConcurrencySerial,
MaxHistorySize: &maxHistory,
})}
auth.bus.On("UserCreatedEvent", "on_user_created", func(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("auth-ok:%s", event.Payload["user_id"]), nil
}, nil)
billing.bus.On("UserCreatedEvent", "on_user_created", func(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("billing-ok:%s", event.Payload["user_id"]), nil
}, nil)
auth.bus.OnEventName("*", "forward_to_relay", func(event *abxbus.BaseEvent) (any, error) {
return relay.bus.Emit(event), nil
}, nil)
relay.bus.OnEventName("*", "forward_to_billing", func(event *abxbus.BaseEvent) (any, error) {
return billing.bus.Emit(event), nil
}, nil)
event := auth.bus.Emit(abxbus.NewBaseEvent("UserCreatedEvent", map[string]any{"user_id": "u-a8d1"}))
result, err := event.EventResult()
if err != nil {
panic(err)
}
fmt.Println(result)
// auth-ok:u-a8d1
fmt.Println(event.EventPath)
// [AuthBus#a8d1 RelayBus#3f2c BillingBus#b91e]
}
```
## Uni-directional and bi-directional forwarding
Forwarding can be one-way or two-way depending on your topology.
* Uni-directional: one producer bus forwards to one consumer bus.
* Bi-directional: both buses forward to each other (common for peer sync).
```python theme={null}
left = EventBus('LeftBus')
right = EventBus('RightBus')
# uni-directional
left.on('*', right.emit)
# bi-directional (add reverse path)
right.on('*', left.emit)
```
```ts theme={null}
const left = new EventBus('LeftBus')
const right = new EventBus('RightBus')
// uni-directional
left.on('*', right.emit)
// bi-directional (add reverse path)
right.on('*', left.emit)
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
event! {
struct PingEvent {
event_result_type: serde_json::Value,
}
}
let left = EventBus::new(Some("LeftBus".to_string()));
let right = EventBus::new(Some("RightBus".to_string()));
let right_for_handler = right.clone();
left.on(PingEvent, move |event: PingEvent| {
let right_for_handler = right_for_handler.clone();
async move {
Ok(right_for_handler.emit(event).to_json_value())
}
});
let left_for_handler = left.clone();
right.on(PingEvent, move |event: PingEvent| {
let left_for_handler = left_for_handler.clone();
async move {
Ok(left_for_handler.emit(event).to_json_value())
}
});
```
```go theme={null}
left := abxbus.NewEventBus("LeftBus", nil)
right := abxbus.NewEventBus("RightBus", nil)
// uni-directional
left.OnEventName("*", "forward_to_right", func(event *abxbus.BaseEvent) (any, error) {
return right.Emit(event), nil
}, nil)
// bi-directional (add reverse path)
right.OnEventName("*", "forward_to_left", func(event *abxbus.BaseEvent) (any, error) {
return left.Emit(event), nil
}, nil)
```
Loop prevention still applies in both modes: if an event already visited a bus (tracked in `event_path`), forwarding back to that bus is a no-op and it is not re-processed there.
## How loop prevention works (`event_path`)
Loop prevention is automatic and based on `event_path`:
1. Each bus appends its own label (for example `AuthBus#a8d1`) to `event_path` when it first sees an event.
2. When a forwarding handler points to another bus, that bus checks whether its label is already in `event_path`.
3. If yes, forwarding to that bus is skipped (no-op), so cycles terminate naturally.
This means you can wire cyclic topologies without infinite forwarding loops.
```python theme={null}
from abxbus import BaseEvent, EventBus
class PingEvent(BaseEvent):
message: str
bus_a = EventBus('BusA')
bus_b = EventBus('BusB')
bus_c = EventBus('BusC')
# cycle: A -> B -> C -> A
bus_a.on('*', bus_b.emit)
bus_b.on('*', bus_c.emit)
bus_c.on('*', bus_a.emit)
event = bus_a.emit(PingEvent(message='hello'))
await event.now()
print(event.event_path)
# ['BusA#a8d1', 'BusB#3f2c', 'BusC#b91e']
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const PingEvent = BaseEvent.extend('PingEvent', {
message: z.string(),
})
const busA = new EventBus('BusA')
const busB = new EventBus('BusB')
const busC = new EventBus('BusC')
// cycle: A -> B -> C -> A
busA.on('*', busB.emit)
busB.on('*', busC.emit)
busC.on('*', busA.emit)
const event = busA.emit(PingEvent({ message: 'hello' }))
await event.now()
console.log(event.event_path)
// ['BusA#a8d1', 'BusB#3f2c', 'BusC#b91e']
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
event! {
struct PingEvent {
event_result_type: serde_json::Value,
}
}
let bus_a = EventBus::new(Some("BusA".to_string()));
let bus_b = EventBus::new(Some("BusB".to_string()));
let bus_c = EventBus::new(Some("BusC".to_string()));
let bus_b_for_handler = bus_b.clone();
bus_a.on(PingEvent, move |event: PingEvent| {
let bus_b_for_handler = bus_b_for_handler.clone();
async move {
Ok(bus_b_for_handler.emit(event).to_json_value())
}
});
let bus_c_for_handler = bus_c.clone();
bus_b.on(PingEvent, move |event: PingEvent| {
let bus_c_for_handler = bus_c_for_handler.clone();
async move {
Ok(bus_c_for_handler.emit(event).to_json_value())
}
});
let bus_a_for_handler = bus_a.clone();
bus_c.on(PingEvent, move |event: PingEvent| {
let bus_a_for_handler = bus_a_for_handler.clone();
async move {
Ok(bus_a_for_handler.emit(event).to_json_value())
}
});
let event = bus_a.emit(PingEvent { ..Default::default() });
block_on(event.now());
println!("{:?}", event.event_path.read());
```
```go theme={null}
busA := abxbus.NewEventBus("BusA", nil)
busB := abxbus.NewEventBus("BusB", nil)
busC := abxbus.NewEventBus("BusC", nil)
// cycle: A -> B -> C -> A
busA.OnEventName("*", "forward_to_b", func(event *abxbus.BaseEvent) (any, error) {
return busB.Emit(event), nil
}, nil)
busB.OnEventName("*", "forward_to_c", func(event *abxbus.BaseEvent) (any, error) {
return busC.Emit(event), nil
}, nil)
busC.OnEventName("*", "forward_to_a", func(event *abxbus.BaseEvent) (any, error) {
return busA.Emit(event), nil
}, nil)
event := busA.Emit(abxbus.NewBaseEvent("PingEvent", map[string]any{"message": "hello"}))
if _, err := event.Now(); err != nil {
panic(err)
}
fmt.Println(event.EventPath)
// [BusA#a8d1 BusB#3f2c BusC#b91e]
```
## Parent-child tracking across forwarded flows
Parent-child tracking also works across forwarded flows:
* if a forwarded event is handled on a downstream bus and that handler emits a child with `event.emit(...)`, the child still links back to the parent via `event_parent_id`
* nested descendants emitted on downstream buses keep that lineage as they continue through forwarding
* this remains true for both queue-jumped children (`await child.now()`) and normally queued children (emitted but not immediately awaited)
* `bus.emit(...)` inside a handler remains detached top-level work with no parent link
See [Parent-Child Tracking](./parent-child-tracking) for a deeper walkthrough and tree-log example.
See [Immediate Execution (RPC-style)](../concurrency/immediate-execution) for queue-jump execution behavior.
## Bridges are forwarding with transport
Bridges are fundamentally the same forwarding pattern, but with serialization + remote transport in the middle.
See [Bridges Overview](../integrations/bridges) for HTTP/Redis/NATS/Postgres/socket/file-backed bridge options and setup patterns.
# Parent-Child Tracking
Source: https://abxbus.archivebox.io/features/parent-child-tracking
Trace nested event flows with automatic parent-child lineage and tree logs.
When a handler emits another event, AbxBus automatically records lineage so you can understand call chains instead of guessing what triggered what.
Repository example files:
* [`examples/parent_child_tracking.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/parent_child_tracking.py)
* [`abxbus-ts/examples/parent_child_tracking.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/parent_child_tracking.ts)
* [`abxbus-rust/tests/test_eventbus_dispatch_parent_tracking.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_eventbus_dispatch_parent_tracking.rs)
* [`examples/log_tree_demo.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/log_tree_demo.py)
* [`abxbus-ts/examples/log_tree_demo.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/log_tree_demo.ts)
## What gets tracked
* `event_parent_id`: points from child -> parent event
* `event_children`: aggregated list of child events emitted with `event.emit(...)` during handler execution
* `event_emitted_by_handler_id`: which specific handler emitted the child
This tracking works across nested chains (parent -> child -> grandchild) and is surfaced in event helpers and tree logs.
## Which emit style should I use?
Most handler code should use `await event.emit(ChildEvent(...)).now()` in Python/TypeScript, `child.Now()` in Go, or `child.now().await` in Rust. That is the "owned child work" style: the child is linked to the parent, runs immediately from the caller's point of view, and the parent cannot complete until the child completes.
Use the other styles only when you explicitly want linked background work or a fully detached top-level event.
| Style | `event_parent_id` | `event_blocks_parent_completion` | Blocks current handler? | Effect |
| - | - | - | - | - |
| `await event.emit(ChildEvent(...)).now()` | Parent event id | `True` / `true` | Yes | Linked child work. The child can queue-jump, and parent completion waits for the child. |
| `event.emit(ChildEvent(...))` without awaiting | Parent event id | `False` / `false` | No | Linked background child. It appears in ancestry/tree logs, but the parent can complete before the child finishes. |
| `await bus.emit(TopLevelEvent(...)).now()` | `None` / `null` | `False` / `false` because there is no linked parent to block | Yes | Detached top-level event that the current handler waits for naturally because it is awaited. It is not a child in hierarchy tracking. |
| `bus.emit(TopLevelEvent(...))` without awaiting | `None` / `null` | `False` / `false` | No | True background event. It is queued for later processing and has no retained relationship to the event that emitted it. |
In short: use `event.emit(...)` when you want parent-child lineage, and add `await` when that child work is part of the parent's completion. Use `bus.emit(...)` when the event should be treated as a separate top-level event.
In Rust, the same split is explicit:
* `current_bus.emit_child(ChildEvent { ..., ..Default::default() })` creates linked child work while inside a handler.
* `child.now().await` queue-jumps and waits for that linked child.
* `bus.emit(EventType { ... })` creates detached top-level work.
* calling `now().await` on a bus-emitted event waits for it without adding parent-child lineage.
In Go, the same split is explicit:
* `child := event.Emit(ChildEvent{...})` creates linked child work.
* `child.Now()` queue-jumps and waits for that linked child.
* `bus.Emit(ParentEvent{...})` creates detached top-level work.
* calling `Now()` on a bus-emitted event waits for it without adding parent-child lineage.
## Works across forwarded buses too
Parent-child lineage is preserved even when the parent event has been forwarded between buses.
If a forwarded event is handled on another bus and that handler emits a child:
* the child still gets `event_parent_id = `
* the child is linked under the emitting handler's `event_children`
* forwarding that child onward keeps the same lineage metadata
Use `event.emit(...)` in handlers so the runtime can attach ancestry and ownership correctly.
See also: [Forwarding Between Buses](./forwarding-between-buses)
## Queue-jumped vs normally queued linked children
Lineage tracking works in both execution styles:
* Queue-jumped child events:
* emitted inside a handler and immediately awaited (`await child.now()`)
* child may execute right away (RPC-style), gets normal parent linkage metadata, and sets `event_blocks_parent_completion=True`
* Normally queued child events:
* emitted inside a handler but not immediately awaited
* child runs later via normal queue scheduling, keeps the same `event_parent_id` ancestry link, and leaves `event_blocks_parent_completion=False`
In short: `event.emit(...)` records lineage. Awaiting the emitted event decides whether that lineage also blocks parent completion.
See [Immediate Execution (RPC-style)](../concurrency/immediate-execution) for queue-jump behavior details.
## Full example: checkout -> reserve/charge/receipt (+ fraud grandchild)
```python theme={null}
from abxbus import BaseEvent, EventBus
class CheckoutEvent(BaseEvent[str]):
order_id: str
class ReserveInventoryEvent(BaseEvent[str]):
order_id: str
class ChargeCardEvent(BaseEvent[str]):
order_id: str
class FraudCheckEvent(BaseEvent[str]):
order_id: str
class SendReceiptEvent(BaseEvent[str]):
order_id: str
bus = EventBus('TreeBus')
async def on_checkout(event: CheckoutEvent) -> str:
reserve = event.emit(ReserveInventoryEvent(order_id=event.order_id))
await reserve.now()
reserve_id = await reserve.event_result()
charge = event.emit(ChargeCardEvent(order_id=event.order_id))
await charge.now()
charge_id = await charge.event_result()
receipt = event.emit(SendReceiptEvent(order_id=event.order_id))
await receipt.now()
receipt_id = await receipt.event_result()
return f'{reserve_id}|{charge_id}|{receipt_id}'
async def on_reserve(event: ReserveInventoryEvent) -> str:
return f'reserve:{event.order_id}'
async def on_charge(event: ChargeCardEvent) -> str:
fraud = event.emit(FraudCheckEvent(order_id=event.order_id))
await fraud.now()
fraud_status = await fraud.event_result()
return f'charge:{event.order_id}:{fraud_status}'
async def on_fraud(event: FraudCheckEvent) -> str:
return f'fraud-ok:{event.order_id}'
async def on_receipt(event: SendReceiptEvent) -> str:
return f'receipt:{event.order_id}'
bus.on(CheckoutEvent, on_checkout)
bus.on(ReserveInventoryEvent, on_reserve)
bus.on(ChargeCardEvent, on_charge)
bus.on(FraudCheckEvent, on_fraud)
bus.on(SendReceiptEvent, on_receipt)
root = bus.emit(CheckoutEvent(order_id='ord-123'))
result = await root.event_result()
await bus.wait_until_idle()
print(result)
print(bus.log_tree())
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const CheckoutEvent = BaseEvent.extend('CheckoutEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
const ReserveInventoryEvent = BaseEvent.extend('ReserveInventoryEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
const ChargeCardEvent = BaseEvent.extend('ChargeCardEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
const FraudCheckEvent = BaseEvent.extend('FraudCheckEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
const SendReceiptEvent = BaseEvent.extend('SendReceiptEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
const bus = new EventBus('TreeBus')
bus.on(CheckoutEvent, async (event) => {
const reserve = event.emit(ReserveInventoryEvent({ order_id: event.order_id }))
await reserve.now()
const charge = event.emit(ChargeCardEvent({ order_id: event.order_id }))
await charge.now()
const receipt = event.emit(SendReceiptEvent({ order_id: event.order_id }))
await receipt.now()
return `${await reserve.eventResult()}|${await charge.eventResult()}|${await receipt.eventResult()}`
})
bus.on(ReserveInventoryEvent, async (event) => `reserve:${event.order_id}`)
bus.on(ChargeCardEvent, async (event) => {
const fraud = event.emit(FraudCheckEvent({ order_id: event.order_id }))
await fraud.now()
return `charge:${event.order_id}:${await fraud.eventResult()}`
})
bus.on(FraudCheckEvent, async (event) => `fraud-ok:${event.order_id}`)
bus.on(SendReceiptEvent, async (event) => `receipt:${event.order_id}`)
const root = bus.emit(CheckoutEvent({ order_id: 'ord-123' }))
await root.now()
await bus.waitUntilIdle()
console.log(await root.eventResult())
console.log(bus.logTree())
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
use serde_json::json;
event! {
struct CheckoutEvent {
order_id: String,
event_result_type: serde_json::Value,
}
}
event! {
struct ReserveInventoryEvent {
order_id: String,
event_result_type: serde_json::Value,
}
}
event! {
struct SendReceiptEvent {
order_id: String,
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("TreeBus".to_string()));
bus.on(CheckoutEvent, |event: CheckoutEvent| async move {
let current_bus = event.event_bus().expect("handler bus");
let order_id = event.order_id;
let reserve = current_bus.emit_child(ReserveInventoryEvent {
order_id: order_id.clone(),
..Default::default()
});
reserve.now().await;
let reserve_id = reserve.event_result().await?;
let receipt = current_bus.emit_child(SendReceiptEvent {
order_id,
..Default::default()
});
receipt.now().await;
let receipt_id = receipt.event_result().await?;
Ok(json!(format!("{reserve_id}|{receipt_id}")))
});
bus.on(ReserveInventoryEvent, |event: ReserveInventoryEvent| async move {
Ok(json!(format!("reserve:{}", event.order_id)))
});
bus.on(SendReceiptEvent, |event: SendReceiptEvent| async move {
Ok(json!(format!("receipt:{}", event.order_id)))
});
let root = bus.emit(CheckoutEvent {
order_id: "ord-123".to_string(),
..Default::default()
});
let result = block_on(root.event_result())?;
block_on(bus.wait_until_idle(None));
println!("{result}");
println!("{}", bus.log_tree());
```
```go theme={null}
package main
import (
"fmt"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
bus := abxbus.NewEventBus("TreeBus", nil)
bus.On("CheckoutEvent", "on_checkout", func(event *abxbus.BaseEvent) (any, error) {
orderID := event.Payload["order_id"].(string)
reserve := event.Emit(abxbus.NewBaseEvent("ReserveInventoryEvent", map[string]any{"order_id": orderID}))
if _, err := reserve.Now(); err != nil {
return nil, err
}
reserveID, err := reserve.EventResult()
if err != nil {
return nil, err
}
charge := event.Emit(abxbus.NewBaseEvent("ChargeCardEvent", map[string]any{"order_id": orderID}))
if _, err := charge.Now(); err != nil {
return nil, err
}
chargeID, err := charge.EventResult()
if err != nil {
return nil, err
}
receipt := event.Emit(abxbus.NewBaseEvent("SendReceiptEvent", map[string]any{"order_id": orderID}))
if _, err := receipt.Now(); err != nil {
return nil, err
}
receiptID, err := receipt.EventResult()
if err != nil {
return nil, err
}
return fmt.Sprintf("%v|%v|%v", reserveID, chargeID, receiptID), nil
}, nil)
bus.On("ReserveInventoryEvent", "on_reserve", func(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("reserve:%s", event.Payload["order_id"]), nil
}, nil)
bus.On("ChargeCardEvent", "on_charge", func(event *abxbus.BaseEvent) (any, error) {
orderID := event.Payload["order_id"].(string)
fraud := event.Emit(abxbus.NewBaseEvent("FraudCheckEvent", map[string]any{"order_id": orderID}))
if _, err := fraud.Now(); err != nil {
return nil, err
}
fraudStatus, err := fraud.EventResult()
if err != nil {
return nil, err
}
return fmt.Sprintf("charge:%s:%v", orderID, fraudStatus), nil
}, nil)
bus.On("FraudCheckEvent", "on_fraud", func(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("fraud-ok:%s", event.Payload["order_id"]), nil
}, nil)
bus.On("SendReceiptEvent", "on_receipt", func(event *abxbus.BaseEvent) (any, error) {
return fmt.Sprintf("receipt:%s", event.Payload["order_id"]), nil
}, nil)
root := bus.Emit(abxbus.NewBaseEvent("CheckoutEvent", map[string]any{"order_id": "ord-123"}))
result, err := root.EventResult()
if err != nil {
panic(err)
}
bus.WaitUntilIdle(nil)
fmt.Println(result)
fmt.Println(bus.LogTree())
}
```
## Example tree output
Captured from running the Python example above with `uv run` (IDs/timestamps vary run-to-run):
```text theme={null}
└── CheckoutEvent#b7c7 [10:10:54.522 (0.003s)]
└── ✅ TreeBus#ef2a.__main__.on_checkout#7a12 [10:10:54.522 (0.002s)] → 'reserve:ord-123|charge:ord-123:fraud-ok:ord-123|receipt:ord-123'
├── ReserveInventoryEvent#ca2f [10:10:54.522 (0.000s)]
│ └── ✅ TreeBus#ef2a.__main__.on_reserve#1583 [10:10:54.522 (0.000s)] → 'reserve:ord-123'
├── ChargeCardEvent#b746 [10:10:54.523 (0.001s)]
│ └── ✅ TreeBus#ef2a.__main__.on_charge#7d9c [10:10:54.523 (0.001s)] → 'charge:ord-123:fraud-ok:ord-123'
│ └── FraudCheckEvent#31e0 [10:10:54.523 (0.000s)]
│ └── ✅ TreeBus#ef2a.__main__.on_fraud#4c4e [10:10:54.523 (0.000s)] → 'fraud-ok:ord-123'
└── SendReceiptEvent#c399 [10:10:54.524 (0.000s)]
└── ✅ TreeBus#ef2a.__main__.on_receipt#de9f [10:10:54.524 (0.000s)] → 'receipt:ord-123'
```
## Why this helps in practice
* Debugging: quickly see causality chains instead of inspecting raw logs line-by-line.
* Reliability: timeout/cancellation behavior can be reasoned about by ancestry.
* Querying: combine lineage with `find(..., child_of=...)` to isolate event families.
## Related pages
* [Immediate Execution (RPC-style)](../concurrency/immediate-execution)
* [Forwarding Between Buses](./forwarding-between-buses)
* [OtelTracingMiddleware](../integrations/middleware-otel-tracing)
* [Find Events](./find-events)
* [BaseEvent](../api/baseevent)
# Return Value Handling
Source: https://abxbus.archivebox.io/features/return-value-handling
Define typed handler returns and collect results from one emitted event.
Handler return values are captured in `EventResult` records and can be consumed as a single value or aggregated across handlers.
Repository example files:
* [`examples/simple.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/simple.py)
* [`abxbus-ts/examples/simple.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/simple.ts)
* [`abxbus-rust/tests/test_event_result.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_event_result.rs)
## Typed return values
Use the event result type to enforce return typing across handlers.
```python theme={null}
from abxbus import BaseEvent, EventBus
class DoMathEvent(BaseEvent[int]):
a: int
b: int
def add(event: DoMathEvent) -> int:
return event.a + event.b
bus = EventBus('AppBus')
bus.on(DoMathEvent, add)
event = await bus.emit(DoMathEvent(a=2, b=3)).now()
result = await event.event_result()
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const DoMathEvent = BaseEvent.extend('DoMathEvent', {
a: z.number(),
b: z.number(),
event_result_type: z.number(),
})
const bus = new EventBus('AppBus')
bus.on(DoMathEvent, (event) => event.a + event.b)
const event = bus.emit(DoMathEvent({ a: 2, b: 3 }))
await event.now()
const result = await event.eventResult()
```
```rust theme={null}
use abxbus::{
event,
event_bus::EventBus,
BaseEvent,
};
use futures::executor::block_on;
event! {
struct DoMathEvent {
a: i64,
b: i64,
event_result_type: i64,
}
}
let bus = EventBus::new(Some("AppBus".to_string()));
bus.on(DoMathEvent, |event: DoMathEvent| async move {
Ok(event.a + event.b)
});
let event = bus.emit(DoMathEvent { a: 2, b: 3, ..Default::default() });
let result = block_on(event.event_result())?;
```
```go theme={null}
type DoMathEvent struct {
A int `json:"a"`
B int `json:"b"`
}
bus := abxbus.NewEventBus("AppBus", nil)
bus.On(func(event DoMathEvent) (int, error) {
return event.A + event.B, nil
})
emitted := bus.Emit(DoMathEvent{A: 2, B: 3})
result, err := emitted.EventResult()
```
## Aggregating multiple handler results
When multiple handlers respond to the same event, collect all successful values with the list helpers.
```python theme={null}
event = await bus.emit(GetConfigEvent()).now()
values = await event.event_results_list(raise_if_any=False, raise_if_none=False)
```
```ts theme={null}
const event = bus.emit(GetConfigEvent({}))
const values = await event.eventResultsList({ raise_if_any: false, raise_if_none: false })
```
```rust theme={null}
use abxbus::{
base_event::EventResultOptions,
event,
event_bus::EventBus,
BaseEvent,
};
use futures::executor::block_on;
use serde_json::json;
event! {
struct GetConfigEvent {
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("ConfigBus".to_string()));
bus.on(GetConfigEvent, |_event: GetConfigEvent| async move {
Ok(json!({"debug": true, "port": 5797}))
});
bus.on(GetConfigEvent, |_event: GetConfigEvent| async move {
Ok(json!({"debug": false, "timeout": 30}))
});
let event = bus.emit(GetConfigEvent { ..Default::default() });
let values = block_on(event.event_results_list_with_options(EventResultOptions {
raise_if_any: false,
raise_if_none: false,
..EventResultOptions::default()
}))?;
```
```go theme={null}
type GetConfigEvent struct{}
event := bus.Emit(GetConfigEvent{})
values, err := event.EventResultsList(
&abxbus.EventResultOptions{RaiseIfAny: false, RaiseIfNone: false},
)
```
## Result list options
```python theme={null}
await event.event_results_list(raise_if_any=False, raise_if_none=False)
await event.event_results_list(include=lambda result, _: isinstance(result, dict), raise_if_any=False)
completed = await event.wait(timeout=0.25)
await completed.event_results_list()
```
```ts theme={null}
await event.eventResultsList({ raise_if_any: false, raise_if_none: false })
await event.eventResultsList({ include: (result) => typeof result === 'object', raise_if_any: false })
await event.wait({ timeout: 0.25 }).eventResultsList()
```
```rust theme={null}
use abxbus::base_event::{EventResultOptions, EventWaitOptions};
event.event_results_list_with_options(EventResultOptions {
raise_if_any: false,
raise_if_none: false,
..EventResultOptions::default()
}).await?;
let completed = event.wait_with_options(EventWaitOptions {
timeout: Some(0.25),
..EventWaitOptions::default()
}).await?;
completed.event_results_list().await?;
```
```go theme={null}
event.EventResultsList(&abxbus.EventResultOptions{
RaiseIfAny: false,
RaiseIfNone: false,
})
timeout := 0.25
completed, err := event.Wait(&abxbus.EventWaitOptions{Timeout: &timeout})
if err != nil {
return err
}
completed.EventResultsList()
```
* Default options are `raise_if_any=true` and `raise_if_none=false` in every runtime.
* `raise_if_any`: raise if any handler ended with an error.
* `raise_if_none`: raise only when no handlers returned a valid value after filtering; it does not raise just because one handler returned `None`/`null`/`undefined`.
* If every handler errors, only `raise_if_any=false` plus `raise_if_none=false` suppresses the error and returns no value/empty list; every other option combination raises.
* A single handler error is raised as that error; multiple handler errors are raised as an aggregate/exception-group shape.
* Default filtering includes only completed, successful, non-empty scalar/object/list values and excludes forwarded `BaseEvent` returns.
* Go accepts no args for defaults (`event.EventResult()` / `event.EventResultsList()`) and a single `nil` options value is equivalent to defaults.
## Per-handler inspection
All runtimes keep per-handler result metadata in addition to the aggregate result helpers.
```python theme={null}
by_name = {result.handler_name: result.result for result in event.event_results.values()}
```
```ts theme={null}
const byHandler = event.event_results
```
```rust theme={null}
let by_name: std::collections::HashMap<_, _> = event
.event_results
.read()
.values()
.map(|result| (result.handler.handler_name.clone(), result.result.clone()))
.collect();
```
```go theme={null}
byName := map[string]any{}
for _, result := range event.EventResults {
byName[result.HandlerName] = result.Result
}
```
# Type-Safe Events
Source: https://abxbus.archivebox.io/features/typed-events
Define validated event payloads and event result types.
Events are strongly typed and validated across the supported runtimes.
Repository example files:
* [`examples/simple.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/simple.py)
* [`abxbus-ts/examples/simple.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/simple.ts)
* [`abxbus-rust/tests/test_base_event.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_base_event.rs)
* [`abxbus-go/README.md`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-go/README.md)
```python theme={null}
from typing import Any
from abxbus import BaseEvent
class OrderCreatedEvent(BaseEvent[dict[str, Any]]):
order_id: str
customer_id: str
total_amount: float
```
```ts theme={null}
import { BaseEvent } from 'abxbus'
import { z } from 'zod'
const OrderCreatedEvent = BaseEvent.extend('OrderCreatedEvent', {
order_id: z.string(),
customer_id: z.string(),
total_amount: z.number(),
event_result_type: z.object({ ok: z.boolean() }),
})
```
```rust theme={null}
use abxbus::{
event,
event_bus::EventBus,
BaseEvent,
};
use futures::executor::block_on;
use serde::{Deserialize, Serialize};
#[derive(Clone, Serialize, Deserialize, PartialEq, Debug)]
struct OrderResult {
ok: bool,
}
event! {
struct OrderCreatedEvent {
order_id: String,
customer_id: String,
total_amount: f64,
event_result_type: OrderResult,
event_result_schema: r#"{
"type": "object",
"properties": {"ok": {"type": "boolean"}},
"required": ["ok"]
}"#,
}
}
let bus = EventBus::new(Some("OrdersBus".to_string()));
bus.on(OrderCreatedEvent, |event: OrderCreatedEvent| async move {
Ok(OrderResult {
ok: event.total_amount > 0.0,
})
});
let event = bus.emit(OrderCreatedEvent {
order_id: "order-123".to_string(),
customer_id: "customer-456".to_string(),
total_amount: 42.50,
..Default::default()
});
let typed_result = block_on(event.event_result())?
.expect("handler result");
```
```go theme={null}
type OrderCreatedEvent struct {
OrderID string `json:"order_id"`
CustomerID string `json:"customer_id"`
TotalAmount float64 `json:"total_amount"`
}
type OrderResult struct {
OK bool `json:"ok"`
}
bus := abxbus.NewEventBus("OrdersBus", nil)
bus.On(func(event OrderCreatedEvent) (OrderResult, error) {
return OrderResult{OK: event.TotalAmount > 0}, nil
})
result, err := bus.Emit(OrderCreatedEvent{
OrderID: "order-123",
CustomerID: "customer-456",
TotalAmount: 42.50,
}).EventResult()
if err != nil {
panic(err)
}
typedResult, err := abxbus.EventResultAs[OrderResult](result)
if err != nil {
panic(err)
}
_ = typedResult
```
* Python payload validation is powered by Pydantic models.
* TypeScript payload and result validation is powered by Zod schemas.
* Rust uses `event!` to keep event fields and event metadata in one block. Primitive result schemas are derived automatically; structured result schemas are supplied through `event_result_schema`.
* Go uses typed struct helpers for static shape at call sites, validates typed handler payloads against the declared payload struct, and derives JSON Schema from the declared result type for runtime result validation.
# Don't Like Events?
Source: https://abxbus.archivebox.io/further-reading/events-suck
Practical patterns for teams who like events in theory but dislike event-driven DX in practice.
If you like events in theory but hate the day-to-day developer experience, this page is for you.
Common pain points:
* calling boilerplate (`emit` + await completion + unwrap result) for every request
* eventual consistency anxiety ("will my response event arrive?")
* duplicating signatures across schemas, handlers, and implementation functions
The goal here is to keep event architecture benefits without forcing painful calling patterns.
## 1) Pain: painful calling interface boilerplate
You usually end up writing verbose call sites repeatedly.
`events_suck.wrap(...)` gives you a method-shaped client API (`client.create(...)`) while still routing through events.
```python theme={null}
# Without wrap: valid, but noisy at every call site
event = bus.emit(CreateUserEvent(name='bob', age=45))
user_id = await event.event_result()
all_values = await event.event_results_list(raise_if_any=False, raise_if_none=False)
# With wrap: looks like normal async function calls
SDKClient = events_suck.wrap('SDKClient', {'create': CreateUserEvent, 'update': UpdateUserEvent})
client = SDKClient(bus=bus)
user_id = await client.create(name='bob', age=45, nickname='bobby')
updated = await client.update(id=user_id, age=46, source='sync')
```
```ts theme={null}
// Without wrap: valid, but noisy at every call site
const event = bus.emit(CreateUserEvent({ name: 'bob', age: 45 }))
await event.now()
const user_id = await event.eventResult()
const all_values = await event.eventResultsList({ raise_if_any: false, raise_if_none: false })
// With wrap: looks like normal async function calls
const SDKClient = events_suck.wrap('SDKClient', { create: CreateUserEvent, update: UpdateUserEvent })
const client = new SDKClient(bus)
const id = await client.create({ name: 'bob', age: 45 }, { nickname: 'bobby' })
const updated = await client.update({ id: id ?? 'fallback-id', age: 46 }, { source: 'sync' })
```
```go theme={null}
// events_suck is not implemented in abxbus-go yet.
//
// Use the core EventBus APIs directly:
event := bus.Emit(abxbus.NewBaseEvent("CreateUserEvent", map[string]any{
"name": "bob",
"age": 45,
}))
userID, err := event.EventResult()
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
### Minimal end-to-end `wrap(...)` wiring
```python theme={null}
from abxbus import BaseEvent, EventBus, events_suck
class CreateUserEvent(BaseEvent[str]):
name: str
age: int
class UpdateUserEvent(BaseEvent[bool]):
id: str
age: int | None = None
class UserService:
def __init__(self) -> None:
self.users: dict[str, dict[str, int | str]] = {}
async def on_create(self, event: CreateUserEvent) -> str:
user_id = f'user-{event.age}'
self.users[user_id] = {'id': user_id, 'name': event.name, 'age': event.age}
return user_id
async def on_update(self, event: UpdateUserEvent) -> bool:
if event.id not in self.users:
return False
if event.age is not None:
self.users[event.id]['age'] = event.age
return True
bus = EventBus('SDKBus')
service = UserService()
bus.on(CreateUserEvent, service.on_create)
bus.on(UpdateUserEvent, service.on_update)
SDKClient = events_suck.wrap('SDKClient', {
'create': CreateUserEvent,
'update': UpdateUserEvent,
})
client = SDKClient(bus=bus)
user_id = await client.create(name='bob', age=45, nickname='bobby')
updated = await client.update(id=user_id, age=46, source='sync')
```
```ts theme={null}
import { BaseEvent, EventBus, events_suck } from 'abxbus'
import { z } from 'zod'
const CreateUserEvent = BaseEvent.extend('CreateUserEvent', {
name: z.string(),
age: z.number(),
event_result_type: z.string(),
})
const UpdateUserEvent = BaseEvent.extend('UpdateUserEvent', {
id: z.string(),
age: z.number().nullable().optional(),
event_result_type: z.boolean(),
})
type UserRecord = { id: string; name: string; age: number }
const users = new Map()
const onCreate = async (event: InstanceType) => {
const user_id = `user-${event.age}`
users.set(user_id, { id: user_id, name: event.name, age: event.age })
return user_id
}
const onUpdate = async (event: InstanceType) => {
const existing = users.get(event.id)
if (!existing) return false
if (event.age !== undefined && event.age !== null) existing.age = event.age
users.set(event.id, existing)
return true
}
const bus = new EventBus('SDKBus')
bus.on(CreateUserEvent, onCreate)
bus.on(UpdateUserEvent, onUpdate)
const SDKClient = events_suck.wrap('SDKClient', {
create: CreateUserEvent,
update: UpdateUserEvent,
})
const client = new SDKClient(bus)
const user_id = await client.create({ name: 'bob', age: 45 }, { nickname: 'bobby' })
const updated = await client.update({ id: user_id ?? 'fallback-id', age: 46 }, { source: 'sync' })
```
```go theme={null}
// events_suck.wrap/make_events/make_handler are not implemented in Go yet.
// For typed event APIs, use bus.On(func(MyEvent) ...) and bus.Emit(MyEvent{...}).
```
```rust theme={null}
// events_suck.wrap/make_events/make_handler are not implemented in the Rust crate yet.
// Use typed events from the `event!` macro and the core EventBus APIs directly.
```
Related docs:
* [Return Value Handling](../features/return-value-handling)
* [BaseEvent API](../api/baseevent)
## 2) Pain: eventual consistency headaches
If your mental model is "I called something, I need a result now," pure fire-and-forget event flows can feel stressful.
Two patterns reduce that stress:
* request/response on one bus with direct return values (`event_result()` after `now({first_result: true})` / `wait({first_result: true})`)
* immediate execution for nested calls inside handlers (RPC-style queue-jump)
These patterns feel function-like for in-process flows. If you later move a step across process/network boundaries (bridges), treat that edge as eventually consistent again.
Immediate execution docs: [Immediate Execution (RPC-style)](../concurrency/immediate-execution)
### Nested request/response with immediate execution
```python theme={null}
class CheckoutEvent(BaseEvent[str]):
order_id: str
class ChargeCardEvent(BaseEvent[str]):
order_id: str
async def on_checkout(event: CheckoutEvent) -> str:
child = event.emit(ChargeCardEvent(order_id=event.order_id))
await child.now() # immediate path while inside handler
receipt_id = await child.event_result()
return receipt_id
async def on_charge(event: ChargeCardEvent) -> str:
return f'receipt-{event.order_id}'
```
```ts theme={null}
const CheckoutEvent = BaseEvent.extend('CheckoutEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
const ChargeCardEvent = BaseEvent.extend('ChargeCardEvent', {
order_id: z.string(),
event_result_type: z.string(),
})
bus.on(CheckoutEvent, async (event) => {
const child = event.emit(ChargeCardEvent({ order_id: event.order_id }))
await child.now() // immediate path while inside handler
return (await child.eventResult()) ?? 'missing-receipt'
})
bus.on(ChargeCardEvent, async (event) => `receipt-${event.order_id}`)
```
```go theme={null}
bus.On("CheckoutEvent", "on_checkout", func(event *abxbus.BaseEvent) (any, error) {
orderID, _ := event.Payload["order_id"].(string)
child := event.Emit(abxbus.NewBaseEvent("ChargeCardEvent", map[string]any{"order_id": orderID}))
if _, err := child.Now(); err != nil {
return nil, err
}
return child.EventResult()
}, nil)
bus.On("ChargeCardEvent", "on_charge", func(event *abxbus.BaseEvent) (any, error) {
orderID, _ := event.Payload["order_id"].(string)
return "receipt-" + orderID, nil
}, nil)
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
event! {
struct CheckoutEvent {
order_id: String,
event_result_type: String,
}
}
event! {
struct ChargeCardEvent {
order_id: String,
event_result_type: String,
}
}
let bus = EventBus::new(Some("CheckoutBus".to_string()));
bus.on(CheckoutEvent, |event: CheckoutEvent| async move {
let child = event.emit(ChargeCardEvent {
order_id: event.order_id,
..Default::default()
});
let completed = child.now().await?;
Ok(completed.event_result().await?.unwrap_or_else(|| "missing-receipt".to_string()))
});
bus.on(ChargeCardEvent, |event: ChargeCardEvent| async move {
Ok(format!("receipt-{}", event.order_id))
});
```
Related docs:
* [Immediate Execution (RPC-style)](../concurrency/immediate-execution)
* [Timeout Enforcement](../concurrency/timeouts)
* [retry](../api/retry)
## 3) Pain: defining signatures multiple times
You can keep one source of truth for payload shapes and reuse it in implementation code.
### One source of truth examples
Use implementation function signatures as the source of truth, then generate event classes from them.
```python theme={null}
from abxbus import EventBus, events_suck
from pydantic import validate_call
@validate_call
def create_user(id: str | None, name: str, age: int) -> str:
return f'{name}-{age}'
@validate_call
def update_user(id: str, age: int | None = None, **extra) -> bool:
return True
events = events_suck.make_events({
'UserCreateEvent': create_user,
'UserUpdateEvent': update_user,
})
bus = EventBus('LegacyBus')
bus.on(events.UserCreateEvent, events_suck.make_handler(create_user))
bus.on(events.UserUpdateEvent, events_suck.make_handler(update_user))
UserClient = events_suck.wrap('UserClient', {'create': events.UserCreateEvent, 'update': events.UserUpdateEvent})
client = UserClient(bus=bus)
```
Keep the schema as the source of truth, infer implementation input types from it, and reuse the same shape in `BaseEvent.extend(...)`.
```ts theme={null}
import { BaseEvent, EventBus, events_suck } from 'abxbus'
import { z } from 'zod'
const CreateUserInputSchema = z.object({
id: z.string().nullable().optional(),
name: z.string(),
age: z.number(),
})
type CreateUserInput = z.infer
const UserCreateEvent = BaseEvent.extend('UserCreateEvent', {
...CreateUserInputSchema.shape,
event_result_type: z.string(),
})
const bus = new EventBus('LegacyBus')
const create_user = async (input: CreateUserInput): Promise => `${input.name}-${input.age}`
bus.on(UserCreateEvent, ({ id, name, age }) => create_user({ id, name, age }))
const UserClient = events_suck.wrap('UserClient', {
create: UserCreateEvent,
})
const client = new UserClient(bus)
const id = await client.create({ id: null, name: 'bob', age: 45 })
```
Keep event payload fields and handler result types in the same event definition, then let handler signatures use the generated Rust struct directly.
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
event! {
struct UserCreateEvent {
id: Option,
name: String,
age: i64,
event_result_type: String,
}
}
let bus = EventBus::new(Some("LegacyBus".to_string()));
bus.on(UserCreateEvent, |event: UserCreateEvent| async move {
Ok(format!("{}-{}", event.name, event.age))
});
let event = bus.emit(UserCreateEvent {
id: None,
name: "bob".to_string(),
age: 45,
..Default::default()
});
let completed = block_on(event.now())?;
let id = block_on(completed.event_result())?;
```
Related docs:
* [Typed Events](../features/typed-events)
* [BaseEvent API](../api/baseevent)
## Migration playbook
1. Start with `wrap(...)` to clean up call-site boilerplate first.
2. Use immediate execution patterns where you need function-call-like request/response behavior.
3. Consolidate types with `@validate_call` (Python) or `z.infer` (TypeScript) to avoid signature drift.
4. Add timeouts/retry policies where needed, instead of forcing eventual-consistency semantics everywhere.
You do not need to choose between clean DX and events. You can keep method-shaped APIs and adopt event internals incrementally.
# Similar Projects
Source: https://abxbus.archivebox.io/further-reading/similar-projects
Similar projects and licensing details.
* [https://www.cosmicpython.com/book/chapter\_08\_events\_and\_message\_bus.html#message\_bus\_diagram](https://www.cosmicpython.com/book/chapter_08_events_and_message_bus.html#message_bus_diagram) ⭐️
* [https://developer.mozilla.org/en-US/docs/Web/API/EventTarget](https://developer.mozilla.org/en-US/docs/Web/API/EventTarget) ⭐️
* [https://github.com/sindresorhus/emittery](https://github.com/sindresorhus/emittery) ⭐️ (equivalent for JS), [https://github.com/EventEmitter2/EventEmitter2](https://github.com/EventEmitter2/EventEmitter2), [https://github.com/vitaly-t/sub-events](https://github.com/vitaly-t/sub-events)
* [https://github.com/pytest-dev/pluggy](https://github.com/pytest-dev/pluggy) ⭐️
* [https://github.com/teamhide/fastapi-event](https://github.com/teamhide/fastapi-event) ⭐️
* [https://github.com/ethereum/lahja](https://github.com/ethereum/lahja) ⭐️
* [https://github.com/enricostara/eventure](https://github.com/enricostara/eventure) ⭐️
* [https://github.com/akhundMurad/diator](https://github.com/akhundMurad/diator) ⭐️
* [https://github.com/n89nanda/pyeventbus](https://github.com/n89nanda/pyeventbus)
* [https://github.com/iunary/aioemit](https://github.com/iunary/aioemit)
* [https://github.com/dboslee/evently](https://github.com/dboslee/evently)
* [https://github.com/faust-streaming/faust](https://github.com/faust-streaming/faust)
* [https://github.com/ArcletProject/Letoderea](https://github.com/ArcletProject/Letoderea)
* [https://github.com/seanpar203/event-bus](https://github.com/seanpar203/event-bus)
* [https://github.com/n89nanda/pyeventbus](https://github.com/n89nanda/pyeventbus)
* [https://github.com/nicolaszein/py-async-bus](https://github.com/nicolaszein/py-async-bus)
* [https://github.com/AngusWG/simple-event-bus](https://github.com/AngusWG/simple-event-bus)
* [https://www.joeltok.com/posts/2021-03-building-an-event-bus-in-python/](https://www.joeltok.com/posts/2021-03-building-an-event-bus-in-python/)
## Distributed Event Queues
* [NATS](https://nats.io/)
* [Kafka](https://kafka.apache.org/)
* [RQ](https://python-rq.org/)
* [Celery](https://docs.celeryq.dev/)
* [Dramatiq](https://dramatiq.io/)
* [Huey](https://huey.readthedocs.io/)
* [RabbitMQ](https://www.rabbitmq.com/)
***
> [🧠 DeepWiki Docs](https://deepwiki.com/ArchiveBox/abxbus)
>
> ![image]()
>
> ![image]()
This project is licensed under the MIT License.
# Overview
Source: https://abxbus.archivebox.io/index
Unified docs for abxbus Python, TypeScript, Rust, and Go implementations.
## `abxbus`: Production-ready multi-language event bus
![abxbus logo]()
[](https://deepwiki.com/ArchiveBox/abxbus) [](https://pypi.org/project/abxbus/) [](https://www.npmjs.com/package/abxbus) [](https://github.com/ArchiveBox/abxbus)
AbxBus is an in-memory event bus for async Python, TypeScript (Node and browser environments), Rust, and Go, built for predictable event-driven workflows with strong typing and consistent cross-language behavior.
Core strengths:
* Typed event payloads and typed handler return values
* Deterministic queue semantics with configurable concurrency
* Nested event lineage tracking (`event_parent_id` / `event_path`)
* Event forwarding, bridge transports, and middleware integration
## Minimal usage
```python theme={null}
from abxbus import BaseEvent, EventBus
class SomeEvent(BaseEvent):
some_data: int
async def on_some_event(event: SomeEvent) -> None:
print(event.some_data)
# 132
bus = EventBus('MyBus')
bus.on(SomeEvent, on_some_event)
await bus.emit(SomeEvent(some_data=132)).now()
```
```ts theme={null}
import { BaseEvent, EventBus } from "abxbus";
import { z } from "zod";
const SomeEvent = BaseEvent.extend("SomeEvent", {
some_data: z.number(),
});
const bus = new EventBus("MyBus");
bus.on(SomeEvent, async (event) => {
console.log(event.some_data);
});
await bus.emit(SomeEvent({ some_data: 132 })).now();
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
event! {
struct SomeEvent {
some_data: i64,
event_result_type: String,
}
}
let bus = EventBus::new(Some("MyBus".to_string()));
bus.on(SomeEvent, |event: SomeEvent| async move {
Ok(format!("did something with {}", event.some_data))
});
let event = bus.emit(SomeEvent { some_data: 132, ..Default::default() });
block_on(event.now())?;
```
```go theme={null}
package main
import (
"fmt"
abxbus "github.com/ArchiveBox/abxbus/abxbus-go"
)
func main() {
type SomeEvent struct {
SomeData int `json:"some_data"`
}
bus := abxbus.NewEventBus("MyBus", nil)
bus.On(func(event SomeEvent) {
fmt.Println(event.SomeData)
})
event := bus.Emit(SomeEvent{SomeData: 132})
_, _ = event.Now()
}
```
See [Quickstart](./quickstart) for installation and first full example.
## Repository examples
Runnable end-to-end examples and implementation smoke tests live in the repo:
* Quickstart basics:
* [`examples/simple.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/simple.py)
* [`abxbus-ts/examples/simple.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/simple.ts)
* Concurrency modes, overrides, and timeout behavior:
* [`examples/concurrency_options.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/concurrency_options.py)
* [`abxbus-ts/examples/concurrency_options.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/concurrency_options.ts)
* Immediate execution (queue-jump) behavior:
* [`examples/immediate_event_processing.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/immediate_event_processing.py)
* [`abxbus-ts/examples/immediate_event_processing.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/immediate_event_processing.ts)
* Forwarding between buses:
* [`examples/forwarding_between_busses.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/forwarding_between_busses.py)
* [`abxbus-ts/examples/forwarding_between_busses.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/forwarding_between_busses.ts)
* Parent-child lineage tracking:
* [`examples/parent_child_tracking.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/parent_child_tracking.py)
* [`abxbus-ts/examples/parent_child_tracking.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/parent_child_tracking.ts)
* Tree logs with nested timeout outcomes:
* [`examples/log_tree_demo.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/log_tree_demo.py)
* [`abxbus-ts/examples/log_tree_demo.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/log_tree_demo.ts)
* Rust parity and roundtrip tests:
* [`abxbus-rust/tests/test_comprehensive_patterns.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_comprehensive_patterns.rs)
* [`abxbus-rust/tests/test_cross_runtime_roundtrip.rs`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/tests/test_cross_runtime_roundtrip.rs)
* Go parity and roundtrip helper:
* [`abxbus-go/tests/comprehensive_patterns_test.go`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-go/tests/comprehensive_patterns_test.go)
* [`abxbus-go/tests/roundtrip_cli`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-go/tests/roundtrip_cli/main.go)
# HTTPEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-http
Forward events over HTTP(S) endpoints.
`HTTPEventBridge` forwards event JSON over HTTP and can optionally expose an inbound HTTP listener.
## Go support
`HTTPEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Constructor params
* `send_to`: optional outbound endpoint (`http://` or `https://`)
* `listen_on`: optional inbound endpoint (`http://` only)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import HTTPEventBridge
bridge = HTTPEventBridge(
send_to='https://peer.example.com/abxbus_events',
listen_on='http://0.0.0.0:8002/abxbus_events',
name='HttpBridge',
)
```
```ts theme={null}
import { HTTPEventBridge } from 'abxbus/bridges'
const bridge = new HTTPEventBridge({
send_to: 'https://peer.example.com/abxbus_events',
listen_on: 'http://0.0.0.0:8002/abxbus_events',
name: 'HttpBridge',
})
```
```go theme={null}
// HTTPEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import HTTPEventBridge
bus = EventBus('AppBus')
bridge = HTTPEventBridge(
send_to='https://peer.example.com/abxbus_events',
listen_on='http://0.0.0.0:8002/abxbus_events',
)
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { HTTPEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new HTTPEventBridge({
send_to: 'https://peer.example.com/abxbus_events',
listen_on: 'http://0.0.0.0:8002/abxbus_events',
})
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
// HTTPEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` serializes an event and sends a `POST` request to `send_to`.
* `on(...)` registers handlers on the bridge's internal inbound bus and auto-starts the listener when needed.
* Inbound payloads are parsed back into `BaseEvent`, reset to pending state, then emitted on the internal bus.
* `close()` shuts down listener/server resources and the internal bus.
* In TypeScript, listener mode (`listen_on`) is supported in Node.js runtimes.
# JSONLEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-jsonl
Forward events through newline-delimited JSON files.
`JSONLEventBridge` appends one event JSON payload per line and tails the file for inbound events.
## Constructor params
* `path`: JSONL file path
* `poll_interval`: tail polling interval in seconds (default `0.25`)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import JSONLEventBridge
bridge = JSONLEventBridge(
'/tmp/abxbus_events.jsonl',
poll_interval=0.25,
name='JsonlBridge',
)
```
```ts theme={null}
import { JSONLEventBridge } from 'abxbus/bridges'
const bridge = new JSONLEventBridge('/tmp/abxbus_events.jsonl', 0.25, 'JsonlBridge')
```
```go theme={null}
bridge := abxbus.NewJSONLEventBridge("/tmp/abxbus_events.jsonl", 0.25, "JsonlBridge")
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import JSONLEventBridge
bus = EventBus('AppBus')
bridge = JSONLEventBridge('/tmp/abxbus_events.jsonl')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { JSONLEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new JSONLEventBridge('/tmp/abxbus_events.jsonl')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
bus := abxbus.NewEventBus("AppBus", nil)
bridge := abxbus.NewJSONLEventBridge("/tmp/abxbus_events.jsonl", 0.25, "JsonlBridge")
bus.OnEventName("*", "jsonl_emit", func(event *abxbus.BaseEvent) (any, error) {
return bridge.Emit(event)
}, nil)
bridge.OnEventName("*", "bus_emit", func(event *abxbus.BaseEvent) (any, error) {
return bus.Emit(event), nil
}, nil)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` appends compact JSON payload + newline to the file.
* `on(...)` auto-starts a tail loop and registers inbound handlers.
* Start cursor is initialized at current EOF, so only newly appended lines are processed.
* Malformed lines are ignored; valid lines are parsed into events, reset, and emitted on the internal bus.
* Runtime note: TypeScript JSONL bridge is Node.js-only. Go's JSONL bridge uses local filesystem polling.
# NATSEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-nats
Forward events over NATS subjects.
`NATSEventBridge` publishes events to a NATS subject and subscribes to the same subject for inbound forwarding.
## Go support
`NATSEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Optional dependencies
Install the NATS extra (recommended):
```bash theme={null}
pip install "abxbus[nats]"
```
Equivalent direct dependency install:
```bash theme={null}
pip install nats-py
```
Install the NATS client package:
```bash theme={null}
npm install abxbus nats
```
This bridge is Node.js-only.
```go theme={null}
// NATSEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for the currently supported Go bridge.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Constructor params
* `server`: NATS server URL (for example `nats://localhost:4222`)
* `subject`: subject name used for publish/subscribe
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import NATSEventBridge
bridge = NATSEventBridge(
'nats://localhost:4222',
'abxbus_events',
name='NatsBridge',
)
```
```ts theme={null}
import { NATSEventBridge } from 'abxbus/bridges'
const bridge = new NATSEventBridge(
'nats://localhost:4222',
'abxbus_events',
'NatsBridge'
)
```
```go theme={null}
// NATSEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import NATSEventBridge
bus = EventBus('AppBus')
bridge = NATSEventBridge('nats://localhost:4222', 'abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { NATSEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new NATSEventBridge('nats://localhost:4222', 'abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
// NATSEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` publishes serialized event JSON bytes to the configured subject.
* `on(...)` registers inbound handlers and auto-starts subscription.
* Inbound messages are decoded, reset, and re-emitted on the internal bus.
* `close()` drains/closes NATS connections and stops the internal bus.
* Runtime requirements: Python needs `nats-py`, TypeScript needs `nats` and Node.js.
# PostgresEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-postgres
Forward events using PostgreSQL LISTEN/NOTIFY plus table storage.
`PostgresEventBridge` stores event payloads in a Postgres table and uses `LISTEN/NOTIFY` for low-latency fanout.
## Go support
`PostgresEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Optional dependencies
Install the Postgres extra (recommended):
```bash theme={null}
pip install "abxbus[postgres]"
```
Equivalent direct dependency install:
```bash theme={null}
pip install asyncpg
```
Install the Postgres client package:
```bash theme={null}
npm install abxbus pg
```
This bridge is Node.js-only.
```go theme={null}
// PostgresEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for the currently supported Go bridge.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Constructor params
* `table_url`: `postgresql://user:pass@host:5432/dbname[/tablename]?...`
* `channel`: optional notify/listen channel (defaults to `abxbus_events`)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import PostgresEventBridge
bridge = PostgresEventBridge(
'postgresql://user:pass@localhost:5432/mydb/abxbus_events',
channel='abxbus_events',
name='PgBridge',
)
```
```ts theme={null}
import { PostgresEventBridge } from 'abxbus/bridges'
const bridge = new PostgresEventBridge(
'postgresql://user:pass@localhost:5432/mydb/abxbus_events',
'abxbus_events',
'PgBridge'
)
```
```go theme={null}
// PostgresEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import PostgresEventBridge
bus = EventBus('AppBus')
bridge = PostgresEventBridge('postgresql://user:pass@localhost:5432/mydb/abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { PostgresEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new PostgresEventBridge('postgresql://user:pass@localhost:5432/mydb/abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
// PostgresEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` upserts event payload data into the bridge table, then sends `NOTIFY` with the event id.
* `on(...)` registers inbound handlers and auto-starts listener startup.
* On notifications, the bridge fetches the full row payload, reconstructs an event, resets it, and emits locally.
* Columns are created on demand only for `event_*` fields.
* The full event JSON is stored in `event_payload` (not just non-`event_*` fields).
* `event_*` columns are also stored as flat mirrored fields for indexing/querying.
* Rehydration merges `event_payload` with `event_*` column values back into a flat event object.
* Runtime requirements: Python needs `asyncpg`, TypeScript needs `pg` and Node.js.
# RedisEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-redis
Forward events via Redis pub/sub channels.
`RedisEventBridge` publishes event payloads to a Redis channel and subscribes for inbound events on the same channel.
## Go support
`RedisEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Optional dependencies
Install the Redis extra (recommended):
```bash theme={null}
pip install "abxbus[redis]"
```
Equivalent direct dependency install:
```bash theme={null}
pip install redis
```
Install the Redis client package:
```bash theme={null}
npm install abxbus ioredis
```
This bridge is Node.js-only.
```go theme={null}
// RedisEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for the currently supported Go bridge.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Constructor params
* `redis_url`: redis URL in the form `redis://user:pass@host:6379//`
* `channel`: optional channel override (defaults to URL channel segment or `abxbus_events`)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import RedisEventBridge
bridge = RedisEventBridge(
'redis://user:pass@localhost:6379/1/abxbus_events',
name='RedisBridge',
)
```
```ts theme={null}
import { RedisEventBridge } from 'abxbus/bridges'
const bridge = new RedisEventBridge(
'redis://user:pass@localhost:6379/1/abxbus_events',
undefined,
'RedisBridge'
)
```
```go theme={null}
// RedisEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import RedisEventBridge
bus = EventBus('AppBus')
bridge = RedisEventBridge('redis://localhost:6379/0/abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { RedisEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new RedisEventBridge('redis://localhost:6379/0/abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
// RedisEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` publishes serialized event JSON to the configured Redis channel.
* `on(...)` subscribes handlers for inbound messages and auto-starts the Redis subscriber.
* Incoming messages are parsed into events, reset, then emitted on the bridge's internal bus.
* `close()` unsubscribes and closes Redis clients.
* Runtime requirements: Python needs `redis` (`redis.asyncio`), TypeScript needs `ioredis` and Node.js.
# SocketEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-socket
Forward events through Unix domain sockets.
`SocketEventBridge` uses a Unix socket path for both send and listen directions.
## Go support
`SocketEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Constructor params
* `path`: Unix socket path (absolute path recommended)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import SocketEventBridge
bridge = SocketEventBridge('/tmp/abxbus_events.sock', name='SocketBridge')
```
```ts theme={null}
import { SocketEventBridge } from 'abxbus/bridges'
const bridge = new SocketEventBridge('/tmp/abxbus_events.sock', 'SocketBridge')
```
```go theme={null}
// SocketEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import SocketEventBridge
bus = EventBus('AppBus')
bridge = SocketEventBridge('/tmp/abxbus_events.sock')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { SocketEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new SocketEventBridge('/tmp/abxbus_events.sock')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
// SocketEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` writes newline-delimited event JSON frames to the Unix socket.
* `on(...)` subscribes handlers on the inbound side and auto-starts the socket listener.
* Incoming frames are decoded into events, reset, then emitted on the internal bus.
* `close()` stops the socket server and tears down the internal bus.
* TypeScript socket bridges require Node.js runtime support for Unix sockets.
# SQLiteEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-sqlite
Forward events through a local SQLite table with polling.
`SQLiteEventBridge` writes events into a SQLite table and polls for newly inserted rows.
## Go support
`SQLiteEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Constructor params
* `path`: SQLite database file path
* `table`: table name (default `abxbus_events`)
* `poll_interval`: polling interval in seconds (default `0.25`)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import SQLiteEventBridge
bridge = SQLiteEventBridge(
'/tmp/abxbus_events.sqlite3',
table='abxbus_events',
poll_interval=0.25,
name='SqliteBridge',
)
```
```ts theme={null}
import { SQLiteEventBridge } from 'abxbus/bridges'
const bridge = new SQLiteEventBridge('/tmp/abxbus_events.sqlite3', 'abxbus_events', 0.25, 'SqliteBridge')
```
```go theme={null}
// SQLiteEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import SQLiteEventBridge
bus = EventBus('AppBus')
bridge = SQLiteEventBridge('/tmp/abxbus_events.sqlite3')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { SQLiteEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new SQLiteEventBridge('/tmp/abxbus_events.sqlite3')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
// SQLiteEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* `emit(...)` upserts event payload fields into the configured table.
* `on(...)` auto-starts polling and registers handlers on the internal bus.
* Columns are created on demand only for `event_*` fields.
* The full event JSON is stored in `event_payload` (not just non-`event_*` fields).
* `event_*` columns are also stored as flat mirrored fields for indexing/querying.
* Writes use SQLite JSON functions for `event_payload` (for example `json(?)`).
* Rehydration merges `event_payload` with `event_*` column values back into a flat event object.
* Rows are read in `(event_created_at, event_id)` order, converted back to events, reset, and emitted locally.
* Runtime notes: Python uses stdlib `sqlite3`; TypeScript requires Node.js with built-in `node:sqlite` (Node 22+).
# TachyonEventBridge
Source: https://abxbus.archivebox.io/integrations/bridge-tachyon
Same-machine SPSC IPC bridge over a shared-memory ring buffer.
`TachyonEventBridge` forwards events between processes on the same machine via a single-producer/single-consumer shared-memory ring buffer (Unix-socket handshake, futex-backed wait). Tachyon's SPSC handshake pairs one producer with one consumer, so deployments typically use one bridge instance for sending and a separate one for receiving; calling both `emit()` and `on()` on the same instance creates a self-loop on the same path.
## Go support
`TachyonEventBridge` is not implemented in `abxbus-go` yet. Use [JSONLEventBridge](./bridge-jsonl) for the currently supported Go bridge.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Optional dependencies
The native Tachyon core requires GCC 14+ or Clang 17+ at install time, so it's exposed as its own extra and intentionally left out of the `bridges` aggregate (which would otherwise fail to install on stock environments). Install it explicitly when you want this bridge.
```bash theme={null}
pip install "abxbus[tachyon]"
```
Equivalent direct install:
```bash theme={null}
pip install tachyon-ipc
```
```bash theme={null}
npm install @tachyon-ipc/core
```
This bridge is Node.js-only. Blocking `listen`/`recv`/`connect` calls run inside `node:worker_threads` so the main event loop stays responsive.
```go theme={null}
// TachyonEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for the currently supported Go bridge.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Constructor params
* `path`: Unix socket path used for the SHM handshake (e.g. `/tmp/abxbus.sock`)
* `capacity`: ring buffer size in bytes; must be a positive power of two (default `1 << 20`)
* `name`: optional bridge label
```python theme={null}
from abxbus.bridges import TachyonEventBridge
bridge = TachyonEventBridge('/tmp/abxbus.sock', capacity=1 << 20, name='TachyonBridge')
```
```ts theme={null}
import { TachyonEventBridge } from 'abxbus/bridges'
const bridge = new TachyonEventBridge('/tmp/abxbus.sock', 1 << 20, 'TachyonBridge')
```
```go theme={null}
// TachyonEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Setup with a bus
The listener side must call `on(...)` before any sender connects so the SHM arena/socket exist on disk.
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import TachyonEventBridge
bus = EventBus('AppBus')
# listener process
inbound = TachyonEventBridge('/tmp/abxbus.sock')
inbound.on('*', bus.emit)
# sender process (separate instance / process)
outbound = TachyonEventBridge('/tmp/abxbus.sock')
bus.on('*', outbound.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { TachyonEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
// listener process
const inbound = new TachyonEventBridge('/tmp/abxbus.sock')
inbound.on('*', bus.emit)
// sender process (separate instance / process)
const outbound = new TachyonEventBridge('/tmp/abxbus.sock')
bus.on('*', outbound.emit)
```
```go theme={null}
// TachyonEventBridge is not implemented in abxbus-go yet.
// Use JSONLEventBridge for Go process-to-process bridge tests and deployments.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed events plus EventBus JSON serialization with an external transport for now.
```
## Behavior
* Tachyon is SPSC: a single producer is paired with a single consumer per session. Use a separate bridge instance per direction if you need bidirectional forwarding.
* The producer disconnects from the Unix socket immediately after the SHM handshake; reconnecting after the consumer has accepted is not supported by Tachyon.
* `close()` on the producer side flushes a reserved `SHUTDOWN_TYPE_ID` sentinel before tearing down its bus so the consumer can exit its blocking `recv` loop without an orphaned thread; the same shutdown wire-format is used in both runtimes, so a Python producer can cleanly shut down a TS listener and vice versa. Only the listener side unlinks the socket on disk.
* Runtime requirements: a Linux/macOS host with futex/`__ulock_wait` support; Python needs `tachyon-ipc`, TypeScript needs `@tachyon-ipc/core` and Node.js.
# Overview
Source: https://abxbus.archivebox.io/integrations/bridges
Transport bridges for forwarding events across files, sockets, and external services.
Bridges are optional connectors for forwarding serialized events between buses in different processes or machines.
All bridges expose the same minimal surface:
* `emit(...)` for outbound forwarding
* `on(...)` for inbound subscription
* `start()` and `close()` for lifecycle control
## Quick setup
```python theme={null}
from abxbus import EventBus
from abxbus.bridges import RedisEventBridge
bus = EventBus('AppBus')
bridge = RedisEventBridge('redis://localhost:6379/0/abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { RedisEventBridge } from 'abxbus/bridges'
const bus = new EventBus('AppBus')
const bridge = new RedisEventBridge('redis://localhost:6379/0/abxbus_events')
bus.on('*', bridge.emit)
bridge.on('*', bus.emit)
```
```go theme={null}
bus := abxbus.NewEventBus("AppBus", nil)
bridge := abxbus.NewJSONLEventBridge("/tmp/abxbus_events.jsonl", 0.25, "JsonlBridge")
bus.OnEventName("*", "jsonl_emit", func(event *abxbus.BaseEvent) (any, error) {
return bridge.Emit(event)
}, nil)
bridge.OnEventName("*", "bus_emit", func(event *abxbus.BaseEvent) (any, error) {
return bus.Emit(event), nil
}, nil)
```
Go currently implements `JSONLEventBridge`. The HTTP, socket, Redis, NATS, Postgres, SQLite, and Tachyon bridges are Python/TypeScript-only for now.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Bridge pages
* [HTTPEventBridge](./bridge-http)
* [SocketEventBridge](./bridge-socket)
* [RedisEventBridge](./bridge-redis)
* [NATSEventBridge](./bridge-nats)
* [PostgresEventBridge](./bridge-postgres)
* [JSONLEventBridge](./bridge-jsonl)
* [SQLiteEventBridge](./bridge-sqlite)
* [TachyonEventBridge](./bridge-tachyon)
# AutoErrorEventMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-auto-error
Emit auto error events when handlers fail.
`AutoErrorEventMiddleware` emits `{OriginalEventType}ErrorEvent` when a handler completes with an error.
## Constructor params
None.
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import AutoErrorEventMiddleware
bus = EventBus(
name='AppBus',
middlewares=[AutoErrorEventMiddleware()],
)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use explicit typed handler wrappers around bus.on(...) for this behavior for now.
```
## Behavior
* Runs on completed handler results.
* If a handler errored, emits an auto event with:
* `event_type`: `{OriginalEventType}ErrorEvent`
* `error`: original exception
* `error_type`: exception class name
* Skips source events ending in `ErrorEvent` or `ResultEvent` to prevent auto recursion.
# AutoHandlerChangeEventMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-auto-handler-change
Emit auto events when handlers are registered/unregistered.
`AutoHandlerChangeEventMiddleware` emits metadata events whenever `EventBus.on(...)` or `EventBus.off(...)` changes handler registration.
## Constructor params
None.
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import AutoHandlerChangeEventMiddleware
bus = EventBus(
name='AppBus',
middlewares=[AutoHandlerChangeEventMiddleware()],
)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use explicit typed handler wrappers around bus.on(...) for this behavior for now.
```
## Behavior
* On registration, emits `BusHandlerRegisteredEvent(handler=...)`.
* On unregistration, emits `BusHandlerUnregisteredEvent(handler=...)`.
* Emits a deep-copied handler metadata object.
* Useful for auditing dynamic handler topology.
# AutoReturnEventMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-auto-return
Emit auto result events for non-None handler returns.
`AutoReturnEventMiddleware` emits `{OriginalEventType}ResultEvent` for successful non-`None` handler return values.
## Constructor params
None.
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import AutoReturnEventMiddleware
bus = EventBus(
name='AppBus',
middlewares=[AutoReturnEventMiddleware()],
)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use explicit typed handler wrappers around bus.on(...) for this behavior for now.
```
## Behavior
* Runs on completed handler results.
* Emits auto result events only when:
* handler returned a non-`None` value
* handler did not error
* return value is not itself a `BaseEvent`
* Auto event payload uses `data=`.
* Skips source events ending in `ErrorEvent` or `ResultEvent`.
# LoggerEventBusMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-logger
Log completed events to stdout and optional file.
`LoggerEventBusMiddleware` prints completed event summaries and can also write them to disk.
## Constructor params
* `log_path`: optional filesystem path for log output
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import LoggerEventBusMiddleware
bus = EventBus(
name='AppBus',
middlewares=[LoggerEventBusMiddleware('./events.log')],
)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use explicit typed handler wrappers around bus.on(...) for this behavior for now.
```
## Behavior
* Logs event summaries when events complete.
* Always prints to stdout.
* If `log_path` is provided, appends the same summary lines to the file.
* Creates parent directories for the log file automatically.
# OtelTracingMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-otel-tracing
Emit OpenTelemetry spans for events and handlers.
`OtelTracingMiddleware` creates event and handler spans with parent-child linking.
## Constructor params
* `tracer` / `Tracer`: optional explicit OpenTelemetry tracer instance.
* `trace_api`: optional explicit `opentelemetry.trace` module in Python/TypeScript.
* `root_span_attributes` / `RootSpanAttributes`: attributes added to top-level event spans.
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import OtelTracingMiddleware
bus = EventBus(
name='AppBus',
middlewares=[OtelTracingMiddleware()],
)
```
```ts theme={null}
import { EventBus } from 'abxbus'
import { OtelTracingMiddleware } from 'abxbus/OtelTracingMiddleware'
const bus = new EventBus('AppBus', {
middlewares: [new OtelTracingMiddleware()],
})
```
```go theme={null}
// Not implemented in the core Go module.
// OpenTelemetry is intentionally not a required Go dependency.
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Setup with Sentry
```python theme={null}
import sentry_sdk
from opentelemetry import trace
from opentelemetry.sdk.trace import TracerProvider
from sentry_sdk.integrations.opentelemetry import SentrySpanProcessor
from abxbus import EventBus
from abxbus.middlewares import OtelTracingMiddleware
sentry_sdk.init(
dsn='https://@/',
traces_sample_rate=1.0,
)
provider = TracerProvider()
provider.add_span_processor(SentrySpanProcessor())
trace.set_tracer_provider(provider)
bus = EventBus(
name='AppBus',
middlewares=[OtelTracingMiddleware()],
)
```
Install requirements:
```bash theme={null}
pip install sentry-sdk opentelemetry-api opentelemetry-sdk
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Wire tracing around typed handlers directly until Rust middleware parity lands.
```
## Behavior
* Starts an event span when an event starts and ends it on completion.
* Starts one child span per handler execution.
* Records handler exceptions on handler spans.
* Links child events to parent handler spans where available.
* With Sentry OpenTelemetry integration enabled, these spans are exported to Sentry performance traces.
# SQLiteHistoryMirrorMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-sqlite-history-mirror
Mirror event and handler snapshots into SQLite tables.
`SQLiteHistoryMirrorMiddleware` records event and handler-result snapshots into SQLite for queryable audit history.
## Constructor params
* `db_path`: SQLite file path
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import SQLiteHistoryMirrorMiddleware
bus = EventBus(
name='AppBus',
middlewares=[SQLiteHistoryMirrorMiddleware('./events.sqlite3')],
)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use explicit typed handler wrappers around bus.on(...) for this behavior for now.
```
## Behavior
* Records event lifecycle snapshots into `events_log`.
* Records handler result snapshots into `event_results_log`.
* Stores serialized payload JSON plus key metadata (event id/type, handler id/name, phase/status).
* Uses WAL mode and thread-safe connection access for concurrent writes.
# WALEventBusMiddleware
Source: https://abxbus.archivebox.io/integrations/middleware-wal
Persist completed events to a JSONL write-ahead log.
`WALEventBusMiddleware` appends completed event snapshots to a JSONL file.
## Constructor params
* `wal_path`: filesystem path for append-only JSONL output
## Setup with EventBus
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import WALEventBusMiddleware
bus = EventBus(
name='AppBus',
middlewares=[WALEventBusMiddleware('./events.jsonl')],
)
```
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use explicit typed handler wrappers around bus.on(...) for this behavior for now.
```
## Behavior
* Writes one JSON line per completed event.
* Uses internal locking for thread-safe file appends.
* Creates parent directories automatically.
* Intended for replay/debug/audit workflows.
# Overview
Source: https://abxbus.archivebox.io/integrations/middlewares
Middleware integrations for EventBus lifecycle hooks across runtimes.
Middlewares can observe event lifecycle transitions, react to handler registration changes, and add cross-cutting behavior around event execution.
## Quick setup
```python theme={null}
from abxbus import EventBus
from abxbus.middlewares import LoggerEventBusMiddleware, WALEventBusMiddleware
bus = EventBus(
name='MyBus',
middlewares=[
WALEventBusMiddleware('./events.jsonl'),
LoggerEventBusMiddleware('./events.log'),
],
)
```
```ts theme={null}
import { BaseEvent, EventBus, type EventStatus } from 'abxbus'
import type { EventBusMiddleware } from 'abxbus/middlewares'
class LoggingMiddleware implements EventBusMiddleware {
async onEventChange(eventbus: EventBus, event: BaseEvent, status: EventStatus): Promise {
if (status === 'completed') {
console.log(`[${eventbus.label}] ${event.event_type}#${event.event_id.slice(-4)}`)
}
}
}
const bus = new EventBus('MyBus', {
middlewares: [LoggingMiddleware],
})
```
```go theme={null}
bus := abxbus.NewEventBus("MyBus", &abxbus.EventBusOptions{
Middlewares: []abxbus.EventBusMiddleware{},
})
```
Go currently ships the middleware interface only. Middleware implementations that require optional third-party packages are not included in the core Go module so those dependencies do not become hard requirements.
```rust theme={null}
// Not implemented in the Rust crate yet.
// Use typed event JSON serialization with an external transport for now.
```
## Middleware interface
Use three optional hooks:
* Event lifecycle: `on_event_change` / `onEventChange`
* Handler-result lifecycle: `on_event_result_change` / `onEventResultChange`
* Handler registration lifecycle: `on_bus_handlers_change` / `onBusHandlersChange`
See [EventBusMiddleware](../api/eventbusmiddleware) for full signatures and custom middleware examples.
## Lifecycle
Middleware hooks receive lifecycle statuses in strict order:
* Event hooks: `pending` -> `started` -> `completed`
* Event-result hooks: `pending` -> `started` -> `completed`
* Handler registration hooks: called when handlers are added or removed via `on(...)` and `off(...)`
`status` passed to lifecycle hooks is never `error`. Handler failures are exposed on `event_result.status` and `event_result.error` during the `completed` callback.
## Built-in classes
* [OtelTracingMiddleware](./middleware-otel-tracing)
* [AutoErrorEventMiddleware](./middleware-auto-error)
* [AutoReturnEventMiddleware](./middleware-auto-return)
* [AutoHandlerChangeEventMiddleware](./middleware-auto-handler-change)
* [WALEventBusMiddleware](./middleware-wal)
* [LoggerEventBusMiddleware](./middleware-logger)
* [SQLiteHistoryMirrorMiddleware](./middleware-sqlite-history-mirror)
* [OtelTracingMiddleware](./middleware-otel-tracing) via `abxbus/OtelTracingMiddleware`
* Middleware interface only.
# Development
Source: https://abxbus.archivebox.io/operations/development
Local development workflows for Python, TypeScript, Rust, and Go.
Set up the Python development environment using `uv`:
```bash theme={null}
git clone https://github.com/ArchiveBox/abxbus && cd abxbus
# Create virtual environment with Python 3.12
uv venv --python 3.12
# Activate virtual environment (varies by OS)
source .venv/bin/activate # On Unix/macOS
# or
.venv\Scripts\activate # On Windows
# Install dependencies
uv sync --dev --all-extras
```
Recommended once per clone:
```bash theme={null}
prek install # install pre-commit hooks
prek run --all-files # run pre-commit hooks on all files manually
```
```bash theme={null}
# Run linter & type checker
uv run ruff check --fix
uv run ruff format
uv run pyright
# Run all tests
uv run pytest -vxs --full-trace tests/
# Run specific test file
uv run pytest tests/test_eventbus.py
# Run Python perf test suite
uv run pytest tests/test_eventbus_performance.py -vxs
# Run the entire lint+test+examples+perf suite for both Python and TS
./test.sh
```
> For AbxBus-TS development see the `abxbus-ts/README.md` `# Development` section.
```bash theme={null}
git clone https://github.com/ArchiveBox/abxbus abxbus && cd abxbus
cd ./abxbus-ts
pnpm install
prek install # install pre-commit hooks
prek run --all-files # run pre-commit hooks on all files manually
pnpm lint
pnpm test
```
```bash theme={null}
git clone https://github.com/ArchiveBox/abxbus abxbus && cd abxbus
cargo test --manifest-path abxbus-rust/Cargo.toml
cargo test --manifest-path abxbus-rust/Cargo.toml --test test_cross_runtime_roundtrip -- --test-threads=1
```
The Rust roundtrip CLI used by cross-runtime tests lives at `abxbus-rust/src/bin/abxbus-rust-roundtrip.rs`.
```bash theme={null}
git clone https://github.com/ArchiveBox/abxbus abxbus && cd abxbus/abxbus-go
go test ./...
go run ./tests/roundtrip_cli events input.json output.json
go run ./tests/roundtrip_cli bus input.json output.json
```
Cross-runtime Go parity is exercised from the repo root by:
```bash theme={null}
cd ..
uv run pytest tests/test_cross_runtime_roundtrip.py -q
pnpm --dir abxbus-ts exec node --expose-gc --test --import tsx tests/cross_runtime_roundtrip.test.ts
```
# Performance
Source: https://abxbus.archivebox.io/operations/performance
Performance notes and benchmark snapshots.
The parity performance suites use the same scenario layout across Python, TypeScript, Rust, and Go:
```bash theme={null}
uv run tests/performance_runtime.py --no-json
pnpm --dir abxbus-ts run perf:node
cargo test --manifest-path abxbus-rust/Cargo.toml --release --test test_eventbus_performance -- --nocapture
(cd abxbus-go && go test ./tests -run TestPerformance -count=1 -timeout=180s -v)
```
Measured locally on the same development machine used for the parity runs:
| Runtime | 1 bus x 50k events x 1 handler | 500 buses x 100 events x 1 handler | 1 bus x 1 event x 50k parallel handlers | 1 bus x 50k events x 50k one-off handlers | Worst case (N buses x N events x N handlers) |
| - | - | - | - | - | - |
| Python | `0.366ms/event`, `0.188kb/event` | `0.408ms/event`, `0.153kb/event` | `0.093ms/handler`, `10.197kb/handler` | `0.633ms/event`, `0.145kb/event` | `0.504ms/event`, `4.171kb/event` |
| Rust | `0.067ms/event` | `0.070ms/event` | `0.062ms/handler` | `0.077ms/event` | `0.227ms/event` |
| Go | `0.016ms/event` | `0.011ms/event` | `0.085ms/handler` | `0.011ms/event` | `0.041ms/event` |
| TypeScript (Node) | `0.065ms/event`, `4.145kb/event` | `0.078ms/event`, `1.562kb/event` | `0.065ms/handler`, `11.631kb/handler` | `0.123ms/event`, `2.182kb/event` | `0.344ms/event`, `12.619kb/event` |
Notes:
* Python and TypeScript memory numbers are peak RSS deltas from their runtime harnesses.
* Rust and Go performance tests currently assert the `0.300ms` per event/handler budget for the measured scenarios but do not report per-event memory in the README matrix.
* Rust's 50k parallel handler budget is measured with `cargo test --release`; the debug build still runs the other Rust performance scenarios and ignores only that release-sensitive stress case.
* The current local Python row and TypeScript worst-case row are above the target budget in some scenarios and should be treated as performance follow-up items rather than baseline targets.
# Supported Runtimes
Source: https://abxbus.archivebox.io/operations/supported-runtimes
Runtime support details for Python, TypeScript, Rust, and Go.
`abxbus` supports Python `3.11+`.
* CPython 3.11 and newer
* OS-independent package support
`abxbus-ts` supports all major JS runtimes.
* Node.js (default development and test runtime)
* Browsers (ESM)
* Bun
* Deno
### Browser support notes
* The package output is ESM (`./dist/esm`) which is supported by all browsers [released after 2018](https://caniuse.com/?search=ESM)
* `AsyncLocalStorage` is preserved at emit time and used during handling when available (Node/Bun), otel/tracing context will work normally in those environments
`abxbus` is an alpha native Rust crate in this repository.
* Tested with Cargo using the crate manifest at `abxbus-rust/Cargo.toml`
* Uses `serde`/`serde_json` for the shared event wire format
* Cross-runtime JSON roundtrips are covered by `abxbus-rust/tests/test_cross_runtime_roundtrip.rs`
`abxbus-go` is an alpha native Go module in this repository.
* Module path: `github.com/ArchiveBox/abxbus/abxbus-go`
* Package import path: `github.com/ArchiveBox/abxbus/abxbus-go`
* Tested with `go test ./...`
* Cross-runtime JSON roundtrips use `abxbus-go/tests/roundtrip_cli`
# Quickstart
Source: https://abxbus.archivebox.io/quickstart
Get started quickly with abxbus in Python, TypeScript, Rust, or Go.
Install abxbus, define one typed event, register a handler, and emit the event. Rust and Go support is currently alpha while cross-runtime parity coverage continues to expand.
Repository example files:
* [`examples/simple.py`](https://github.com/ArchiveBox/abxbus/blob/main/examples/simple.py)
* [`abxbus-ts/examples/simple.ts`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-ts/examples/simple.ts)
* [`abxbus-rust/README.md`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-rust/README.md)
* [`abxbus-go/README.md`](https://github.com/ArchiveBox/abxbus/blob/main/abxbus-go/README.md)
## Install
```bash theme={null}
pip install abxbus
```
```bash theme={null}
npm install abxbus
```
```bash theme={null}
cargo test --manifest-path abxbus-rust/Cargo.toml
```
```bash theme={null}
go get github.com/ArchiveBox/abxbus/abxbus-go
```
## First event
```python theme={null}
import asyncio
from abxbus import BaseEvent, EventBus
class CreateUserEvent(BaseEvent[dict]):
email: str
async def on_create_user(event: CreateUserEvent) -> dict:
user = await your_create_user_logic(event.email)
return {'user_id': user['id']}
async def main() -> None:
bus = EventBus('MyAuthEventBus')
bus.on(CreateUserEvent, on_create_user)
event = await bus.emit(CreateUserEvent(email='someuser@example.com')).now()
result = await event.event_result()
print(result)
# {'user_id': 'some-user-uuid'}
asyncio.run(main())
```
```ts theme={null}
import { BaseEvent, EventBus } from 'abxbus'
import { z } from 'zod'
const CreateUserEvent = BaseEvent.extend('CreateUserEvent', {
email: z.string(),
event_result_type: z.object({ user_id: z.string() }),
})
const bus = new EventBus('MyAuthEventBus')
bus.on(CreateUserEvent, async (event) => {
const user = await yourCreateUserLogic(event.email)
return { user_id: user.id }
})
const event = bus.emit(CreateUserEvent({ email: 'someuser@example.com' }))
await event.now()
console.log(await event.eventResult()) // { user_id: 'some-user-uuid' }
```
```rust theme={null}
use abxbus::{event, event_bus::EventBus};
use futures::executor::block_on;
use serde_json::json;
event! {
struct CreateUserEvent {
email: String,
event_result_type: serde_json::Value,
}
}
let bus = EventBus::new(Some("MyAuthEventBus".to_string()));
bus.on(CreateUserEvent, |event: CreateUserEvent| async move {
let email = event.email.clone();
Ok(json!({"user_id": format!("created:{email}")}))
});
let event = bus.emit(CreateUserEvent {
email: "someuser@example.com".to_string(),
..Default::default()
});
block_on(async {
let completed = event.now().await?;
let result = completed.event_result().await?;
println!("{result:?}");
Ok::<(), String>(())
})?;
```
```go theme={null}
type CreateUserEvent struct {
Email string `json:"email"`
}
type CreateUserResult struct {
UserID string `json:"user_id"`
}
bus := abxbus.NewEventBus("MyAuthEventBus", nil)
bus.On(func(event CreateUserEvent) (CreateUserResult, error) {
return CreateUserResult{UserID: "some-user-uuid"}, nil
})
event := bus.Emit(CreateUserEvent{Email: "someuser@example.com"})
result, err := event.EventResult()
if err != nil {
panic(err)
}
fmt.Println(result)
```
## Next steps
* Browse the [Features](./features/event-pattern-matching) section for behavior patterns.
* Use [API Reference](./api/eventbus) for signatures and options.
* See [Integrations](./integrations/bridges) for bridges and middleware.