Package org.a2aproject.sdk.server.events

Class MainEventBusProcessor

java.lang.Object

org.a2aproject.sdk.server.events.MainEventBusProcessor

All Implemented Interfaces:

Runnable

@ApplicationScoped
public class MainEventBusProcessor
extends Object
implements Runnable

Background processor for the MainEventBus.

This processor runs in a dedicated background thread, consuming events from the MainEventBus and performing two critical operations in order:

1. Update TaskStore with event data (persistence FIRST)
2. Distribute event to ChildQueues (clients see it AFTER persistence)

This architecture ensures clients never receive events before they're persisted, eliminating race conditions and enabling reliable event replay.

Note: This bean is eagerly initialized by MainEventBusProcessorInitializer to ensure the background thread starts automatically when the application starts.

Exception Handling

TaskStore persistence failures are caught and handled gracefully:

• TaskSerializationException - Data corruption or schema mismatch. Logged at ERROR level, distributed as InternalError to clients.
• TaskPersistenceException - Database/storage system failure. Logged at ERROR level, distributed as InternalError to clients.

Processing continues after errors - the failed event is distributed as InternalError to all ChildQueues, and the MainEventBusProcessor continues consuming subsequent events.

Constructor Summary

Constructors

Modifier	Constructor	Description
protected	MainEventBusProcessor()	No-arg constructor for CDI proxying.
	MainEventBusProcessor(MainEventBus eventBus, TaskStore taskStore, PushNotificationSender pushSender, QueueManager queueManager)

Method Summary

Modifier and Type	Method	Description
void	ensureStarted()	No-op method to force CDI proxy resolution and ensure @PostConstruct has been called.
void	run()
void	setCallback(MainEventBusProcessorCallback callback)	Set a callback for testing synchronization with async event processing.
void	setPushNotificationExecutor(Executor executor)	Set a custom executor for push notifications (primarily for testing).

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Constructor Details

MainEventBusProcessor

protected MainEventBusProcessor()

No-arg constructor for CDI proxying. CDI requires this for @ApplicationScoped beans. Fields are initialized via the @Inject constructor.

MainEventBusProcessor

@Inject
public MainEventBusProcessor(MainEventBus eventBus,
 TaskStore taskStore,
 PushNotificationSender pushSender,
 QueueManager queueManager)

Method Details

setCallback

public void setCallback(MainEventBusProcessorCallback callback)

Set a callback for testing synchronization with async event processing.

This is primarily intended for tests that need to wait for event processing to complete. Pass null to reset to the default NOOP callback.

Parameters:

callback - the callback to invoke during event processing, or null for NOOP

setPushNotificationExecutor

public void setPushNotificationExecutor(Executor executor)

Set a custom executor for push notifications (primarily for testing).

By default, push notifications are sent asynchronously using CompletableFuture.runAsync() with the default ForkJoinPool. For tests that need deterministic ordering of push notifications, inject a synchronous executor that runs tasks immediately on the calling thread.

Example synchronous executor for tests:

 Executor syncExecutor = Runnable::run;
 mainEventBusProcessor.setPushNotificationExecutor(syncExecutor);
 

Parameters:

executor - the executor to use for push notifications, or null to use default ForkJoinPool

ensureStarted

public void ensureStarted()

No-op method to force CDI proxy resolution and ensure @PostConstruct has been called. Called by MainEventBusProcessorInitializer during application startup.

run

public void run()

Specified by:

run in interface Runnable