Handlers: serial - abxbus

serial executes handlers for a single event sequentially. Companion runnable example:

Lifecycle impact

Event starts processing.
Handler A runs to completion (or failure/timeout).
Handler B starts afterward, then C, and so on.
Event completion waits for the serial chain (or completion-mode short-circuit rules).

Execution order example

Python
TypeScript
Go
Rust

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']

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

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

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.