Skip to content

Commit 7e73cab

Browse files
014-codekabir
andauthored
fix(server): start event bus processor for manual wiring
Ensure MainEventBusProcessor.ensureStarted() starts manually constructed processors, and call it from DefaultRequestHandler.create(...). This prevents manually wired integrations from leaving events stuck on the main event bus. Add a regression test for the manual wiring path that returns a Message response. Start the main event bus processor from the public request handler constructor and verify the direct manual wiring path. This fixes #995 Co-authored-by: Kabir Khan <kkhan@redhat.com>
1 parent d6068f5 commit 7e73cab

5 files changed

Lines changed: 109 additions & 14 deletions

File tree

‎docs/content/dev/server.md‎

Lines changed: 47 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -247,4 +247,51 @@ See [Backward Compatibility](compatibility) for multi-version modules, version r
247247
- **Quarkus** — Reference implementations are Quarkus-based (JSON-RPC, gRPC, REST)
248248
- **Jakarta EE** — [a2a-jakarta](https://github.com/wildfly-extras/a2a-jakarta) works with any Jakarta EE Web Profile runtime
249249

250+
### Manual Wiring
251+
252+
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:
253+
254+
```java
255+
@Configuration
256+
class A2AServerConfiguration {
257+
258+
@Bean
259+
MainEventBus mainEventBus() {
260+
return new MainEventBus();
261+
}
262+
263+
@Bean
264+
MainEventBusProcessor mainEventBusProcessor(MainEventBus mainEventBus, TaskStore taskStore,
265+
PushNotificationSender pushSender, QueueManager queueManager) {
266+
return new MainEventBusProcessor(mainEventBus, taskStore, pushSender, queueManager);
267+
}
268+
269+
@Bean(destroyMethod = "shutdown")
270+
ExecutorService a2aEventConsumerExecutor() {
271+
return Executors.newCachedThreadPool();
272+
}
273+
274+
@Bean
275+
RequestHandler requestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
276+
QueueManager queueManager, PushNotificationConfigStore pushConfigStore,
277+
MainEventBusProcessor mainEventBusProcessor,
278+
TaskAuthorizationProvider authorizationProvider,
279+
@Qualifier("a2aInternalExecutor") Executor executor,
280+
@Qualifier("a2aEventConsumerExecutor") Executor eventConsumerExecutor) {
281+
return DefaultRequestHandler.builder()
282+
.agentExecutor(agentExecutor)
283+
.taskStore(taskStore)
284+
.queueManager(queueManager)
285+
.pushConfigStore(pushConfigStore)
286+
.mainEventBusProcessor(mainEventBusProcessor)
287+
.authorizationProvider(authorizationProvider)
288+
.executor(executor)
289+
.eventConsumerExecutor(eventConsumerExecutor)
290+
.build();
291+
}
292+
}
293+
```
294+
295+
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.
296+
250297
See [CONTRIBUTING_INTEGRATIONS.md](https://github.com/a2aproject/a2a-java/blob/main/CONTRIBUTING_INTEGRATIONS.md) to submit your own integration.

‎server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessor.java‎

Lines changed: 10 additions & 10 deletions
Original file line numberDiff line numberDiff line change
@@ -156,25 +156,25 @@ int pushNotificationChainCount() {
156156
return pushNotificationChains.size();
157157
}
158158

159+
/**
160+
* Starts the background processor thread if it is not already running.
161+
*/
159162
@SuppressWarnings("NullAway.Init")
160163
@PostConstruct
161-
void start() {
164+
public synchronized void start() {
165+
if (processorThread != null && processorThread.isAlive()) {
166+
LOGGER.debug("MainEventBusProcessor already started");
167+
return;
168+
}
169+
running = true;
162170
processorThread = new Thread(this, "MainEventBusProcessor");
163171
processorThread.setDaemon(true); // Allow JVM to exit even if this thread is running
164172
processorThread.start();
165173
LOGGER.info("MainEventBusProcessor started");
166174
}
167175

168-
/**
169-
* No-op method to force CDI proxy resolution and ensure @PostConstruct has been called.
170-
* Called by MainEventBusProcessorInitializer during application startup.
171-
*/
172-
public void ensureStarted() {
173-
// Method intentionally empty - just forces proxy resolution
174-
}
175-
176176
@PreDestroy
177-
void stop() {
177+
synchronized void stop() {
178178
LOGGER.info("MainEventBusProcessor stopping...");
179179
running = false;
180180
if (processorThread != null) {

‎server-common/src/main/java/org/a2aproject/sdk/server/events/MainEventBusProcessorInitializer.java‎

Lines changed: 2 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -27,14 +27,12 @@ public class MainEventBusProcessorInitializer {
2727

2828
/**
2929
* Observes ApplicationScoped initialization to force eager creation of MainEventBusProcessor.
30-
* The injection of MainEventBusProcessor in this bean triggers its creation, and calling
31-
* ensureStarted() forces the CDI proxy to be resolved, which ensures @PostConstruct has been
30+
* Calling start() forces the CDI proxy to be resolved, which ensures @PostConstruct has been
3231
* called and the background thread is running.
3332
*/
3433
void onStart(@Observes @Initialized(ApplicationScoped.class) Object event) {
3534
if (processor != null) {
36-
// Force proxy resolution to ensure @PostConstruct has been called
37-
processor.ensureStarted();
35+
processor.start();
3836
LOGGER.info("MainEventBusProcessor initialized and started");
3937
} else {
4038
LOGGER.error("MainEventBusProcessor is null - initialization failed!");

‎server-common/src/main/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandler.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -308,6 +308,10 @@ protected DefaultRequestHandler() {
308308
this.eventConsumerExecutor = null;
309309
}
310310

311+
/**
312+
* Constructor used by CDI to create the request handler. Its signature may change between
313+
* releases; application code should use {@link #builder()} to configure and create a handler.
314+
*/
311315
@Inject
312316
public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
313317
QueueManager queueManager, PushNotificationConfigStore pushConfigStore,
@@ -326,6 +330,7 @@ public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
326330
// I am unsure about the correct scope.
327331
// Also reworked to make a Supplier since otherwise the builder gets polluted with wrong tasks
328332
this.requestContextBuilder = () -> new SimpleRequestContextBuilder(taskStore, false, null);
333+
this.mainEventBusProcessor.start();
329334
}
330335

331336
@SuppressWarnings("NullAway.Init")

‎server-common/src/test/java/org/a2aproject/sdk/server/requesthandlers/DefaultRequestHandlerTest.java‎

Lines changed: 45 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -186,6 +186,51 @@ void testInitConfigReadsBlockingTimeouts() {
186186
assertEquals(7, handler.reconciliationTimeoutSeconds);
187187
}
188188

189+
@Test
190+
void testBuilderStartsManuallyConstructedMainEventBusProcessor() throws Exception {
191+
InMemoryTaskStore manualTaskStore = new InMemoryTaskStore();
192+
PushNotificationConfigStore manualPushConfigStore = new InMemoryPushNotificationConfigStore();
193+
MainEventBus manualMainEventBus = new MainEventBus();
194+
InMemoryQueueManager manualQueueManager = new InMemoryQueueManager(manualTaskStore, manualMainEventBus);
195+
MainEventBusProcessor manualProcessor = new MainEventBusProcessor(
196+
manualMainEventBus, manualTaskStore, NOOP_PUSHNOTIFICATION_SENDER, manualQueueManager);
197+
198+
try {
199+
AgentExecutor manualAgentExecutor = new AgentExecutor() {
200+
@Override
201+
public void execute(RequestContext context, AgentEmitter agentEmitter) {
202+
agentEmitter.sendMessage("manual lifecycle response");
203+
}
204+
205+
@Override
206+
public void cancel(RequestContext context, AgentEmitter agentEmitter) {
207+
throw new AssertionError("Cancel should not be invoked");
208+
}
209+
};
210+
DefaultRequestHandler manualRequestHandler = DefaultRequestHandler.builder()
211+
.agentExecutor(manualAgentExecutor)
212+
.taskStore(manualTaskStore)
213+
.queueManager(manualQueueManager)
214+
.pushConfigStore(manualPushConfigStore)
215+
.mainEventBusProcessor(manualProcessor)
216+
.executor(internalExecutor)
217+
.eventConsumerExecutor(internalExecutor)
218+
.authorizationRequired(false)
219+
.build();
220+
manualRequestHandler.agentCompletionTimeoutSeconds = 5;
221+
manualRequestHandler.consumptionCompletionTimeoutSeconds = 2;
222+
manualRequestHandler.reconciliationTimeoutSeconds = 1;
223+
224+
EventKind eventKind = manualRequestHandler.onMessageSend(
225+
MessageSendParams.builder().message(MESSAGE).build(), NULL_CONTEXT);
226+
227+
assertInstanceOf(Message.class, eventKind);
228+
assertEquals(Message.Role.ROLE_AGENT, ((Message) eventKind).role());
229+
} finally {
230+
EventQueueUtil.stop(manualProcessor);
231+
}
232+
}
233+
189234
/**
190235
* Test 1: Non-streaming AUTH_REQUIRED returns immediately while agent continues.
191236
* Verifies:

0 commit comments

Comments
 (0)