diff --git a/docs/content/dev/server.md b/docs/content/dev/server.md index 72edda957..91d1a3e09 100644 --- a/docs/content/dev/server.md +++ b/docs/content/dev/server.md @@ -247,4 +247,51 @@ See [Backward Compatibility](compatibility) for multi-version modules, version r - **Quarkus** — Reference implementations are Quarkus-based (JSON-RPC, gRPC, REST) - **Jakarta EE** — [a2a-jakarta](https://github.com/wildfly-extras/a2a-jakarta) works with any Jakarta EE Web Profile runtime +### Manual Wiring + +For example, when wiring the server components as Spring beans, expose the event processor as a bean and pass it to the request handler. The handler starts the processor when it is constructed, including when the processor was created manually: + +```java +@Configuration +class A2AServerConfiguration { + + @Bean + MainEventBus mainEventBus() { + return new MainEventBus(); + } + + @Bean + MainEventBusProcessor mainEventBusProcessor(MainEventBus mainEventBus, TaskStore taskStore, + PushNotificationSender pushSender, QueueManager queueManager) { + return new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager); + } + + @Bean(destroyMethod = "shutdown") + ExecutorService a2aEventConsumerExecutor() { + return Executors.newCachedThreadPool(); + } + + @Bean + RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore, + QueueManager queueManager, PushNotificationConfigStore pushConfigStore, + MainEventBusProcessor mainEventBusProcessor, + TaskAuthorizationProvider authorizationProvider, + @Qualifier("a2aInternalExecutor") Executor executor, + @Qualifier("a2aEventConsumerExecutor") Executor eventConsumerExecutor) { + return DefaultRequestHandler.builder() + .agentExecutor(agentExecutor) + .taskStore(taskStore) + .queueManager(queueManager) + .pushConfigStore(pushConfigStore) + .mainEventBusProcessor(mainEventBusProcessor) + .authorizationProvider(authorizationProvider) + .executor(executor) + .eventConsumerExecutor(eventConsumerExecutor) + .build(); + } +} +``` + +Register the remaining dependencies, including an `a2aInternalExecutor` bean and a `TaskAuthorizationProvider`, as Spring beans. Keep the cached event consumer executor separate from the agent executor: blocking event polling can exhaust a bounded pool under load. + See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration. diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java index cd293f672..c0bf0e92b 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java @@ -156,25 +156,25 @@ int pushNotificationChainCount() { return pushNotificationChains.size(); } + /** + * Starts the background processor thread if it is not already running. + */ @SuppressWarnings("NullAway.Init") @PostConstruct - void start() { + public synchronized void start() { + if (processorThread != null && processorThread.isAlive()) { + LOGGER.debug("MainEventBusProcessor already started"); + return; + } + running = true; processorThread = new Thread(this, "MainEventBusProcessor"); processorThread.setDaemon(true); // Allow JVM to exit even if this thread is running processorThread.start(); LOGGER.info("MainEventBusProcessor started"); } - /** - * No-op method to force CDI proxy resolution and ensure @PostConstruct has been called. - * Called by MainEventBusProcessorInitializer during application startup. - */ - public void ensureStarted() { - // Method intentionally empty - just forces proxy resolution - } - @PreDestroy - void stop() { + synchronized void stop() { LOGGER.info("MainEventBusProcessor stopping..."); running = false; if (processorThread != null) { diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java index 707b04d1f..615cd4c46 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java @@ -27,14 +27,12 @@ public class MainEventBusProcessorInitializer { /** * Observes ApplicationScoped initialization to force eager creation of MainEventBusProcessor. - * The injection of MainEventBusProcessor in this bean triggers its creation, and calling - * ensureStarted() forces the CDI proxy to be resolved, which ensures @PostConstruct has been + * Calling start() forces the CDI proxy to be resolved, which ensures @PostConstruct has been * called and the background thread is running. */ void onStart(@Observes @Initialized(ApplicationScoped.class) Object event) { if (processor != null) { - // Force proxy resolution to ensure @PostConstruct has been called - processor.ensureStarted(); + processor.start(); LOGGER.info("MainEventBusProcessor initialized and started"); } else { LOGGER.error("MainEventBusProcessor is null - initialization failed!"); diff --git a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java index 7c97cc003..4388b0443 100644 --- a/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java +++ b/server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java @@ -308,6 +308,10 @@ protected DefaultRequestHandler() { this.eventConsumerExecutor = null; } + /** + * Constructor used by CDI to create the request handler. Its signature may change between + * releases; application code should use {@link #builder()} to configure and create a handler. + */ @Inject public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore, QueueManager queueManager, PushNotificationConfigStore pushConfigStore, @@ -326,6 +330,7 @@ public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore, // I am unsure about the correct scope. // Also reworked to make a Supplier since otherwise the builder gets polluted with wrong tasks this.requestContextBuilder = () -> new SimpleRequestContextBuilder(taskStore, false, null); + this.mainEventBusProcessor.start(); } @SuppressWarnings("NullAway.Init") diff --git a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java index 53b25b1a1..1e729bd20 100644 --- a/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java +++ b/server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java @@ -186,6 +186,51 @@ void testInitConfigReadsBlockingTimeouts() { assertEquals(7, handler.reconciliationTimeoutSeconds); } + @Test + void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception { + InMemoryTaskStore manualTaskStore = new InMemoryTaskStore(); + PushNotificationConfigStore manualPushConfigStore = new InMemoryPushNotificationConfigStore(); + MainEventBus manualMainEventBus = new MainEventBus(); + InMemoryQueueManager manualQueueManager = new InMemoryQueueManager(manualTaskStore, manualMainEventBus); + MainEventBusProcessor manualProcessor = new MainEventBusProcessor( + manualMainEventBus, manualTaskStore, NOOP_PUSHNOTIFICATION_SENDER, manualQueueManager); + + try { + AgentExecutor manualAgentExecutor = new AgentExecutor() { + @Override + public void execute(RequestContext context, AgentEmitter agentEmitter) { + agentEmitter.sendMessage("manual lifecycle response"); + } + + @Override + public void cancel(RequestContext context, AgentEmitter agentEmitter) { + throw new AssertionError("Cancel should not be invoked"); + } + }; + DefaultRequestHandler manualRequestHandler = DefaultRequestHandler.builder() + .agentExecutor(manualAgentExecutor) + .taskStore(manualTaskStore) + .queueManager(manualQueueManager) + .pushConfigStore(manualPushConfigStore) + .mainEventBusProcessor(manualProcessor) + .executor(internalExecutor) + .eventConsumerExecutor(internalExecutor) + .authorizationRequired(false) + .build(); + manualRequestHandler.agentCompletionTimeoutSeconds = 5; + manualRequestHandler.consumptionCompletionTimeoutSeconds = 2; + manualRequestHandler.reconciliationTimeoutSeconds = 1; + + EventKind eventKind = manualRequestHandler.onMessageSend( + MessageSendParams.builder().message(MESSAGE).build(), NULL_CONTEXT); + + assertInstanceOf(Message.class, eventKind); + assertEquals(Message.Role.ROLE_AGENT, ((Message) eventKind).role()); + } finally { + EventQueueUtil.stop(manualProcessor); + } + } + /** * Test 1: Non-streaming AUTH_REQUIRED returns immediately while agent continues. * Verifies: