Handler Completion: all

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

package main

import (
	"sync"
	"time"

	abxbus "github.com/ArchiveBox/abxbus/abxbus-go/v2"
)

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

use abxbus_rust::{
    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);