> ## Documentation Index
> Fetch the complete documentation index at: https://abxbus.archivebox.io/llms.txt
> Use this file to discover all available pages before exploring further.

# EventBus

> EventBus constructors, configuration, and core methods.

`EventBus` is the central runtime for handler registration, event emit, history lookup, and lifecycle control.

## `EventBus(...)`

<Tabs>
  <Tab title="Python">
    ```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,
    )
    ```
  </Tab>

  <Tab title="TypeScript">
    ```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 | new () => EventBusMiddleware>
    })
    ```
  </Tab>

  <Tab title="Go">
    ```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
    }
    ```
  </Tab>

  <Tab title="Rust">
    ```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()
        },
    );
    ```
  </Tab>
</Tabs>

### 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 `'*'`).

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    bus.on(UserEvent, handler)
    bus.on('UserEvent', handler)
    bus.on('*', wildcard_handler)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    bus.on(UserEvent, handler)
    bus.on('UserEvent', handler)
    bus.on('*', wildcardHandler)
    ```
  </Tab>

  <Tab title="Go">
    ```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,
    })
    ```
  </Tab>

  <Tab title="Rust">
    ```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) },
    );
    ```
  </Tab>
</Tabs>

## `off(...)`

Unregisters handlers by event key, handler function/reference, or handler id.

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    bus.off(UserEvent, handler)
    bus.off(UserEvent)   # remove all handlers for UserEvent
    bus.off('*')         # remove all wildcard handlers
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    bus.off(UserEvent, handler)
    bus.off(UserEvent)
    bus.off('*')
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    handler := bus.On("UserEvent", "handler", onUserEvent, nil)
    bus.Off("UserEvent", handler)
    bus.Off("UserEvent", handler.ID)
    bus.Off("*", nil)
    ```
  </Tab>

  <Tab title="Rust">
    ```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);
    ```
  </Tab>
</Tabs>

## `emit(...)`

`emit(...)` enqueues synchronously and returns the pending event immediately.

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    event = bus.emit(MyEvent(data='x'))
    result = await event.event_result()
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    const event = bus.emit(MyEvent({ data: 'x' }))
    const result = await event.now({ first_result: true }).eventResult()
    ```
  </Tab>

  <Tab title="Go">
    ```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
    }
    ```
  </Tab>

  <Tab title="Rust">
    ```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())?;
    ```
  </Tab>
</Tabs>

## `find(...)`

`find(...)` supports history lookup, optional future waiting, predicate filtering, and parent/child scoping.

<Tabs>
  <Tab title="Python">
    ```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)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```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 })
    ```
  </Tab>

  <Tab title="Go">
    ```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,
    })
    ```
  </Tab>

  <Tab title="Rust">
    ```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()
        },
    ));
    ```
  </Tab>
</Tabs>

## `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.

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    recent = await bus.filter(ResponseEvent, past=10, future=False, limit=5)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    const recent = await bus.filter(ResponseEvent, { past: 10, future: false, limit: 5 })
    ```
  </Tab>

  <Tab title="Go">
    ```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},
    )
    ```
  </Tab>

  <Tab title="Rust">
    ```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()
        },
    ));
    ```
  </Tab>
</Tabs>

## Lifecycle helpers

### Wait for idle

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    await bus.wait_until_idle()
    await bus.wait_until_idle(timeout=5)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    await bus.waitUntilIdle()
    await bus.waitUntilIdle(5)
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    bus.WaitUntilIdle(nil)
    timeout := 5.0
    bus.WaitUntilIdle(&timeout)
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    use futures::executor::block_on;

    block_on(bus.wait_until_idle(None));
    block_on(bus.wait_until_idle(Some(5.0)));
    ```
  </Tab>
</Tabs>

### Parent/child relationship checks

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    bus.event_is_child_of(child_event, parent_event)
    bus.event_is_parent_of(parent_event, child_event)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    bus.eventIsChildOf(childEvent, parentEvent)
    bus.eventIsParentOf(parentEvent, childEvent)
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    bus.EventIsChildOf(childEvent, parentEvent)
    bus.EventIsParentOf(parentEvent, childEvent)
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    bus.event_is_child_of(&child_event, &parent_event);
    bus.event_is_parent_of(&parent_event, &child_event);
    ```
  </Tab>
</Tabs>

### 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.

<Tabs>
  <Tab title="Python">
    ```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
    ```
  </Tab>

  <Tab title="TypeScript">
    ```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 })
    ```
  </Tab>

  <Tab title="Go">
    ```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()
    ```
  </Tab>

  <Tab title="Rust">
    ```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();
    ```
  </Tab>
</Tabs>

## `destroy(...)`

Destroy stops runtime work and optionally clears bus-owned state.

<Tabs>
  <Tab title="Python">
    ```python theme={null}
    await bus.destroy()
    await bus.destroy(clear=False)
    ```
  </Tab>

  <Tab title="TypeScript">
    ```ts theme={null}
    await bus.destroy()
    await bus.destroy({ clear: false })
    ```
  </Tab>

  <Tab title="Go">
    ```go theme={null}
    bus.Destroy()
    bus.DestroyWithOptions(&abxbus.EventBusDestroyOptions{Clear: false})
    ```
  </Tab>

  <Tab title="Rust">
    ```rust theme={null}
    use abxbus::event_bus::DestroyOptions;

    bus.destroy();
    bus.destroy_with_options(DestroyOptions { clear: false, ..Default::default() });
    ```
  </Tab>
</Tabs>

* `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.