Skip to content

Commit 12bb82e

Browse files
014-codekabir
authored andcommitted
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. This fixes a2aproject#995
1 parent d6068f5 commit 12bb82e

2 files changed

Lines changed: 48 additions & 5 deletions

File tree

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

Lines changed: 13 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -158,23 +158,31 @@ int pushNotificationChainCount() {
158158

159159
@SuppressWarnings("NullAway.Init")
160160
@PostConstruct
161-
void start() {
161+
synchronized void start() {
162+
if (processorThread != null) {
163+
LOGGER.debug("MainEventBusProcessor already started");
164+
return;
165+
}
166+
running = true;
162167
processorThread = new Thread(this, "MainEventBusProcessor");
163168
processorThread.setDaemon(true); // Allow JVM to exit even if this thread is running
164169
processorThread.start();
165170
LOGGER.info("MainEventBusProcessor started");
166171
}
167172

168173
/**
169-
* No-op method to force CDI proxy resolution and ensure @PostConstruct has been called.
170-
* Called by MainEventBusProcessorInitializer during application startup.
174+
* Ensures the background processor thread has been started.
175+
* <p>
176+
* In CDI runtimes, this forces proxy resolution and {@link PostConstruct}. For manual
177+
* wiring, this starts the processor directly.
178+
* </p>
171179
*/
172180
public void ensureStarted() {
173-
// Method intentionally empty - just forces proxy resolution
181+
start();
174182
}
175183

176184
@PreDestroy
177-
void stop() {
185+
synchronized void stop() {
178186
LOGGER.info("MainEventBusProcessor stopping...");
179187
running = false;
180188
if (processorThread != null) {

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

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

189+
@Test
190+
void testCreateStartsManuallyConstructedMainEventBusProcessor() 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+
RequestHandler manualRequestHandler = DefaultRequestHandler.create(
211+
manualAgentExecutor, manualTaskStore, manualQueueManager, manualPushConfigStore,
212+
manualProcessor, internalExecutor, internalExecutor);
213+
214+
EventKind eventKind = manualRequestHandler.onMessageSend(
215+
MessageSendParams.builder().message(MESSAGE).build(), NULL_CONTEXT);
216+
217+
assertInstanceOf(Message.class, eventKind);
218+
assertEquals(Message.Role.ROLE_AGENT, ((Message) eventKind).role());
219+
} finally {
220+
EventQueueUtil.stop(manualProcessor);
221+
}
222+
}
223+
189224
/**
190225
* Test 1: Non-streaming AUTH_REQUIRED returns immediately while agent continues.
191226
* Verifies:

0 commit comments

Comments
 (0)