fix(client): deliver exactly one terminal SSE callback (#1170) by chopmob-cloud · Pull Request #1173 · a2aproject/a2a-java · GitHub
Skip to content

fix(client): deliver exactly one terminal SSE callback (#1170) #1173

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
ehsavoie merged 3 commits into from
Sep 24, 2026
Merged

fix(client): deliver exactly one terminal SSE callback (#1170) #1173

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
81dfe0e
fix(client): deliver exactly one terminal SSE callback (#1170)
chopmob-cloud Sep 23, 2026
29e414a
fix(client): signal completion for final SSE events
kabir Sep 24, 2026
13e4f40
test(client): cover final SSE cancellation callback
kabir Sep 24, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
21 changes: 2 additions & 19 deletions ...onrpc/src/main/java/org/a2aproject/sdk/client/transport/jsonrpc/sse/SSEEventListener.java
View file
Open in desktop
130 changes: 129 additions & 1 deletion ...c/src/test/java/org/a2aproject/sdk/client/transport/jsonrpc/sse/SSEEventListenerTest.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -4,12 +4,18 @@
import static org.junit.jupiter.api.Assertions.assertFalse;
import static org.junit.jupiter.api.Assertions.assertInstanceOf;
import static org.junit.jupiter.api.Assertions.assertNotNull;
import static org.junit.jupiter.api.Assertions.assertNull;
import static org.junit.jupiter.api.Assertions.assertSame;
import static org.junit.jupiter.api.Assertions.assertTrue;

import java.util.concurrent.CountDownLatch;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.TimeoutException;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

Expand Down Expand Up @@ -248,6 +254,128 @@ public void testOnEventWithFinalTaskStatusUpdateEventEventCancels() throws Excep
}



@Test
public void testOnCompleteThenOnErrorDeliversSingleTerminalCallback() {
AtomicInteger terminalCount = new AtomicInteger(0);
AtomicReference<Throwable> lastArg = new AtomicReference<>();
SSEEventListener listener = new SSEEventListener(
event -> {},
error -> { terminalCount.incrementAndGet(); lastArg.set(error); });

listener.onComplete();
listener.onError(new RuntimeException("late error"), new CancelCapturingFuture());

assertEquals(1, terminalCount.get());
assertNull(lastArg.get());
}

@Test
public void testOnErrorThenOnCompleteDeliversSingleTerminalCallback() {
AtomicInteger terminalCount = new AtomicInteger(0);
AtomicReference<Throwable> lastArg = new AtomicReference<>();
SSEEventListener listener = new SSEEventListener(
event -> {},
error -> { terminalCount.incrementAndGet(); lastArg.set(error); });

RuntimeException boom = new RuntimeException("first error");
listener.onError(boom, new CancelCapturingFuture());
listener.onComplete();

assertEquals(1, terminalCount.get());
assertSame(boom, lastArg.get());
}

@Test
public void testRepeatedOnCompleteDeliversSingleTerminalCallback() {
AtomicInteger terminalCount = new AtomicInteger(0);
AtomicReference<Throwable> lastArg = new AtomicReference<>();
SSEEventListener listener = new SSEEventListener(
event -> {},
error -> { terminalCount.incrementAndGet(); lastArg.set(error); });

listener.onComplete();
listener.onComplete();
listener.onComplete();

assertEquals(1, terminalCount.get());
}

@Test
public void testConcurrentTerminalSignalsDeliverExactlyOneCallback() throws Exception {
AtomicInteger terminalCount = new AtomicInteger(0);
AtomicReference<Throwable> lastArg = new AtomicReference<>();
SSEEventListener listener = new SSEEventListener(
event -> {},
error -> { terminalCount.incrementAndGet(); lastArg.set(error); });

int threads = 32;
ExecutorService pool = Executors.newFixedThreadPool(threads);
CountDownLatch start = new CountDownLatch(1);
CountDownLatch done = new CountDownLatch(threads);
for (int i = 0; i < threads; i++) {
final boolean complete = (i % 2 == 0);
pool.submit(() -> {
try {
start.await();
if (complete) {
listener.onComplete();
} else {
listener.onError(new RuntimeException("race"), new CancelCapturingFuture());
}
} catch (InterruptedException e) {
Thread.currentThread().interrupt();
} finally {
done.countDown();
}
});
}
start.countDown();
assertTrue(done.await(10, TimeUnit.SECONDS));
pool.shutdownNow();

assertEquals(1, terminalCount.get());
}

@Test
public void testFinalEventThenOnCompleteDeliversSingleTerminalCallback() throws Exception {
AtomicInteger terminalCount = new AtomicInteger(0);
SSEEventListener listener = new SSEEventListener(
event -> {},
error -> terminalCount.incrementAndGet());

String eventData = JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.substring(
JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.indexOf("{"));
CancelCapturingFuture future = new CancelCapturingFuture();
listener.onMessage(new ServerSentEvent(eventData), future);
listener.onComplete();

assertTrue(future.cancelHandlerCalled);
assertEquals(1, terminalCount.get());
}

@Test
public void testFinalEventThenOnErrorDeliversNormalCompletionOnly() {
AtomicInteger terminalCount = new AtomicInteger(0);
AtomicReference<Throwable> terminalError = new AtomicReference<>();
SSEEventListener listener = new SSEEventListener(
event -> {},
error -> {
terminalCount.incrementAndGet();
terminalError.set(error);
});

String eventData = JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.substring(
JsonStreamingMessages.STREAMING_STATUS_UPDATE_EVENT_FINAL.indexOf("{"));
CancelCapturingFuture future = new CancelCapturingFuture();
listener.onMessage(new ServerSentEvent(eventData), future);
listener.onError(new RuntimeException("cancelled after final event"), future);

assertTrue(future.cancelHandlerCalled);
assertEquals(1, terminalCount.get());
assertNull(terminalError.get());
}

private static class CancelCapturingFuture implements Future<Void> {
private boolean cancelHandlerCalled;

Expand Down Expand Up @@ -280,4 +408,4 @@ public Void get(long timeout, TimeUnit unit) throws InterruptedException, Execut
return null;
}
}
}
}
28 changes: 25 additions & 3 deletions ...i/src/main/java/org/a2aproject/sdk/client/transport/spi/sse/AbstractSSEEventListener.java
View file
Open in desktop
Original file line number Diff line number Diff line change
@@ -1,6 +1,7 @@
package org.a2aproject.sdk.client.transport.spi.sse;

import java.util.concurrent.Future;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.Consumer;
import java.util.logging.Logger;

Expand All @@ -25,6 +26,7 @@ public abstract class AbstractSSEEventListener {

private final Consumer<StreamingEventKind> eventHandler;
private final @Nullable Consumer<Throwable> errorHandler;
private final AtomicBoolean terminalSignaled = new AtomicBoolean(false);

/**
* Creates a new SSE event listener with the specified handlers.
Expand Down Expand Up @@ -73,14 +75,33 @@ protected Consumer<StreamingEventKind> getEventHandler() {
* @param future Optional future for closing the SSE connection
*/
public void onError(Throwable throwable, @Nullable Future<Void> future) {
if (errorHandler != null) {
errorHandler.accept(throwable);
}
signalTerminal(throwable);
if (future != null) {
future.cancel(true); // close SSE channel
}
}

/**
* Delivers exactly one terminal callback for the stream. The first caller to win
* the atomic transition delivers its outcome to the error/completion consumer (a
* non-null {@code error} is a failure, {@code null} is normal completion); every
* later completion, error or post-cancellation signal is dropped, so a single
* streaming request yields exactly one terminal callback.
*
* @param error the failure to report, or {@code null} to signal normal completion
*/
protected void signalTerminal(@Nullable Throwable error) {
if (!terminalSignaled.compareAndSet(false, true)) {
LOGGER.fine("Terminal callback already delivered, ignoring subsequent signal");
return;
}
if (errorHandler != null) {
errorHandler.accept(error);
} else if (error != null) {
LOGGER.warning("errorHandler is null, cannot report terminal error");
}
}

/**
* Processes a parsed streaming event and handles auto-close logic for final events.
* This method encapsulates the common logic for handling events and determining
Expand All @@ -97,6 +118,7 @@ protected void handleEvent(StreamingEventKind event, @Nullable Future<Void> futu
// This covers late subscriptions to completed tasks and ensures no connection leaks
if (shouldAutoClose(event) && future != null) {
LOGGER.fine("Auto-closing SSE connection for final event: " + event.getClass().getSimpleName());
signalTerminal(null);
future.cancel(true); // close SSE channel
}
}
Expand Down
39 changes: 17 additions & 22 deletions .../java/org/a2aproject/sdk/compat03/client/transport/jsonrpc/sse/SSEEventListener_v0_3.java
View file
Open in desktop
Loading
Loading