From ce37b2216f42b5ef8a262f20ba304ee52719da0f Mon Sep 17 00:00:00 2001 From: Emmanuel Hugonnet Date: Fri, 11 Sep 2026 14:49:46 +0200 Subject: [PATCH] feat(sse): add SSEParserConfig, raise default per-line limit to 1 MB - Introduce SSEParserConfig record with builder to make SSE parser limits (maxLineLength, maxBufferLines, maxBufferChars) configurable - Raise default per-line limit from 64 KB to 1 MB so large JSON-RPC responses are no longer silently dropped - Thread SSEParserConfig through all HTTP client implementations (JDK, Android, OkHttp) via new constructors and provider methods - Prevent corrupt/skipped event blocks from advancing lastEventId, which could cause event loss on reconnect - Add AbstractA2AHttpClientSSETest for shared cross-client SSE tests - Document SSEParserConfig usage and security considerations Fixes #1123 Signed-off-by: Emmanuel Hugonnet --- docs/content/dev/client.md | 29 ++ .../http/android/AndroidA2AHttpClient.java | 43 ++- .../android/AndroidA2AHttpClientProvider.java | 15 + .../android/AndroidA2AHttpClientSSETest.java | 6 + .../http/android/OkHttpA2AHttpClient.java | 36 +- .../android/OkHttpA2AHttpClientSSETest.java | 6 + .../sdk/client/http/A2AHttpClientFactory.java | 29 ++ .../client/http/A2AHttpClientProvider.java | 13 + .../sdk/client/http/JdkA2AHttpClient.java | 31 +- .../client/http/JdkA2AHttpClientProvider.java | 8 + .../sdk/client/http/SSEParserConfig.java | 87 +++++ .../client/http/ServerSentEventParser.java | 37 +- .../client/http/A2AHttpClientFactoryTest.java | 18 +- .../http/A2AHttpClientProviderTest.java | 9 + .../http/AbstractA2AHttpClientSSETest.java | 79 +++++ .../client/http/JdkA2AHttpClientSSETest.java | 14 + .../sdk/client/http/JdkA2AHttpClientTest.java | 2 +- .../http/ServerSentEventParserTest.java | 333 +++++++++++++++++- 18 files changed, 767 insertions(+), 28 deletions(-) create mode 100644 http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java create mode 100644 http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java diff --git a/docs/content/dev/client.md b/docs/content/dev/client.md index e1668d873..6c9c858ea 100644 --- a/docs/content/dev/client.md +++ b/docs/content/dev/client.md @@ -185,6 +185,35 @@ Client client = Client .build(); ``` +### SSE Parser Configuration + +The client uses a Server-Sent Events (SSE) parser for streaming responses. You can tune its limits via `SSEParserConfig` to handle agents that return large payloads (e.g., large artifacts or tool results): + +| Parameter | Description | Default | +|------------------|----------------------------------------------------------|-----------| +| `maxLineLength` | Max characters per raw SSE line (`0` = disabled) | 1 MB | +| `maxBufferLines` | Max `data:` lines per event block | 1 000 | +| `maxBufferChars` | Max total characters across all `data:` values per event | 1 MB | + +The defaults are suitable for most deployments. To override them, build a custom `SSEParserConfig` and pass it to the HTTP client. + +> **Security note:** setting `maxLineLength` to `0` disables the per-line length check, relying on `maxBufferChars` alone to bound memory. Only disable it when you trust the remote agent or have other safeguards (e.g. a reverse proxy with its own line-length limit). + +```java +SSEParserConfig sseConfig = SSEParserConfig.builder() + .maxLineLength(4 * 1024 * 1024) // 4 MB per line + .maxBufferChars(4 * 1024 * 1024) // 4 MB per event + .build(); + +// Pass to JdkA2AHttpClient, then to your transport config +JdkA2AHttpClient httpClient = new JdkA2AHttpClient(sseConfig); + +Client client = Client + .builder(agentCard) + .withTransport(JSONRPCTransport.class, new JSONRPCTransportConfig(httpClient)) + .build(); +``` + ## Observability (Optional) Add distributed tracing and W3C Trace Context propagation to client calls with the [OpenTelemetry extras modules](extra/opentelemetry#client). diff --git a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java index c78389620..97973b97e 100644 --- a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java +++ b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClient.java @@ -1,5 +1,7 @@ package org.a2aproject.sdk.client.http.android; +import static org.a2aproject.sdk.util.Assert.checkNotNullParam; + import static java.net.HttpURLConnection.HTTP_FORBIDDEN; import static java.net.HttpURLConnection.HTTP_MULT_CHOICE; import static java.net.HttpURLConnection.HTTP_OK; @@ -28,6 +30,7 @@ import org.a2aproject.sdk.client.http.A2AHttpHeaders; import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.client.http.ServerSentEvent; +import org.a2aproject.sdk.client.http.SSEParserConfig; import org.a2aproject.sdk.client.http.ServerSentEventParser; import org.a2aproject.sdk.common.A2AErrorMessages; import org.a2aproject.sdk.spec.A2AClientHTTPError; @@ -48,24 +51,44 @@ public class AndroidA2AHttpClient implements A2AHttpClient { return t; }); + private final SSEParserConfig sseParserConfig; + + public AndroidA2AHttpClient() { + this(SSEParserConfig.DEFAULT); + } + + /** + * Creates a new Android HTTP client with custom SSE parser limits. + * + * @param sseParserConfig the SSE parser configuration to use for streaming responses + */ + public AndroidA2AHttpClient(SSEParserConfig sseParserConfig) { + this.sseParserConfig = checkNotNullParam("sseParserConfig", sseParserConfig); + } + @Override public GetBuilder createGet() { - return new AndroidGetBuilder(); + return new AndroidGetBuilder(sseParserConfig); } @Override public PostBuilder createPost() { - return new AndroidPostBuilder(); + return new AndroidPostBuilder(sseParserConfig); } @Override public DeleteBuilder createDelete() { - return new AndroidDeleteBuilder(); + return new AndroidDeleteBuilder(sseParserConfig); } private abstract static class AndroidBuilder> implements Builder { protected String url = ""; protected Map headers = new HashMap<>(); + protected final SSEParserConfig sseParserConfig; + + AndroidBuilder(SSEParserConfig sseParserConfig) { + this.sseParserConfig = sseParserConfig; + } @Override public T url(String url) { @@ -203,7 +226,7 @@ protected void processSSEResponse( BufferedReader reader = new BufferedReader(new InputStreamReader(is, StandardCharsets.UTF_8))) { String line; if (isSse) { - ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer); + ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig); while ((line = reader.readLine()) != null) { sseParser.processLine(line); } @@ -244,6 +267,10 @@ protected CompletableFuture executeAsyncSSE( } private static class AndroidGetBuilder extends AndroidBuilder implements GetBuilder { + AndroidGetBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse get() throws IOException { HttpURLConnection connection = createConnection("GET", false); @@ -271,6 +298,10 @@ private static class AndroidPostBuilder extends AndroidBuilder private String body = ""; private boolean followRedirects = false; + AndroidPostBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public PostBuilder body(String body) { this.body = body; @@ -325,6 +356,10 @@ public CompletableFuture postAsyncSSE( private static class AndroidDeleteBuilder extends AndroidBuilder implements DeleteBuilder { + AndroidDeleteBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse delete() throws IOException { HttpURLConnection connection = createConnection("DELETE", false); diff --git a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java index 2cb173385..a59de7164 100644 --- a/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java +++ b/extras/http-client-android/src/main/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientProvider.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.A2AHttpClientProvider; +import org.a2aproject.sdk.client.http.SSEParserConfig; /** * Service provider for {@link AndroidA2AHttpClient}. @@ -24,6 +25,20 @@ public A2AHttpClient create() { return new AndroidA2AHttpClient(); } + /** + * {@inheritDoc} + * + * @throws IllegalStateException if the Android runtime is not available + */ + @Override + public A2AHttpClient create(SSEParserConfig sseParserConfig) { + if (!ANDROID_AVAILABLE) { + throw new IllegalStateException( + "Android classes are not available. This provider is only supported on Android."); + } + return new AndroidA2AHttpClient(sseParserConfig); + } + @Override public int priority() { return ANDROID_AVAILABLE ? 110 : -1; // Higher priority than Vert.x on Android diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java index 963545e85..18b5992c9 100644 --- a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/AndroidA2AHttpClientSSETest.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.AbstractA2AHttpClientSSETest; +import org.a2aproject.sdk.client.http.SSEParserConfig; public class AndroidA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { @@ -9,4 +10,9 @@ public class AndroidA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { protected A2AHttpClient createClient() { return new AndroidA2AHttpClient(); } + + @Override + protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return new AndroidA2AHttpClient(sseParserConfig); + } } diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java index fbe147f87..03b143c9f 100644 --- a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClient.java @@ -22,6 +22,7 @@ import org.a2aproject.sdk.client.http.A2AHttpHeaders; import org.a2aproject.sdk.client.http.A2AHttpResponse; import org.a2aproject.sdk.client.http.ServerSentEvent; +import org.a2aproject.sdk.client.http.SSEParserConfig; import org.a2aproject.sdk.client.http.ServerSentEventParser; import org.a2aproject.sdk.common.A2AErrorMessages; import org.a2aproject.sdk.spec.A2AClientHTTPError; @@ -48,24 +49,39 @@ class OkHttpA2AHttpClient implements A2AHttpClient { return t; }); + private final SSEParserConfig sseParserConfig; + + OkHttpA2AHttpClient() { + this(SSEParserConfig.DEFAULT); + } + + OkHttpA2AHttpClient(SSEParserConfig sseParserConfig) { + this.sseParserConfig = sseParserConfig; + } + @Override public GetBuilder createGet() { - return new OkHttpGetBuilder(); + return new OkHttpGetBuilder(sseParserConfig); } @Override public PostBuilder createPost() { - return new OkHttpPostBuilder(); + return new OkHttpPostBuilder(sseParserConfig); } @Override public DeleteBuilder createDelete() { - return new OkHttpDeleteBuilder(); + return new OkHttpDeleteBuilder(sseParserConfig); } private abstract static class OkHttpBuilder> implements Builder { protected String url = ""; protected final Map headers = new HashMap<>(); + protected final SSEParserConfig sseParserConfig; + + OkHttpBuilder(SSEParserConfig sseParserConfig) { + this.sseParserConfig = sseParserConfig; + } @Override public T url(String url) { @@ -189,7 +205,7 @@ private void parseResponseBody( new InputStreamReader(body.byteStream(), StandardCharsets.UTF_8))) { String line; if (isSse) { - ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer); + ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig); while ((line = reader.readLine()) != null) { sseParser.processLine(line); } @@ -214,6 +230,10 @@ private void parseResponseBody( } private static class OkHttpGetBuilder extends OkHttpBuilder implements GetBuilder { + OkHttpGetBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse get() throws IOException { OkHttpClient client = buildClient(false); @@ -248,6 +268,10 @@ private static class OkHttpPostBuilder extends OkHttpBuilder implem private String body = ""; private boolean followRedirects = false; + OkHttpPostBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public PostBuilder body(String body) { this.body = body; @@ -295,6 +319,10 @@ public CompletableFuture postAsyncSSE( } private static class OkHttpDeleteBuilder extends OkHttpBuilder implements DeleteBuilder { + OkHttpDeleteBuilder(SSEParserConfig sseParserConfig) { + super(sseParserConfig); + } + @Override public A2AHttpResponse delete() throws IOException { OkHttpClient client = buildClient(false); diff --git a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java index c664ca1d3..c672a518a 100644 --- a/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java +++ b/extras/http-client-android/src/test/java/org/a2aproject/sdk/client/http/android/OkHttpA2AHttpClientSSETest.java @@ -2,6 +2,7 @@ import org.a2aproject.sdk.client.http.A2AHttpClient; import org.a2aproject.sdk.client.http.AbstractA2AHttpClientSSETest; +import org.a2aproject.sdk.client.http.SSEParserConfig; public class OkHttpA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { @@ -9,4 +10,9 @@ public class OkHttpA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { protected A2AHttpClient createClient() { return new OkHttpA2AHttpClient(); } + + @Override + protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return new OkHttpA2AHttpClient(sseParserConfig); + } } diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java index 52931720c..fe0b448b6 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientFactory.java @@ -85,6 +85,35 @@ public static A2AHttpClient create() { .orElseThrow(() -> new IllegalStateException("No A2AHttpClientProvider could be instantiated")); } + /** + * Creates a new A2AHttpClient instance with the given {@link SSEParserConfig} using the + * highest available priority provider. + * + *

+ * Providers that support SSE parser configuration will use it; others fall back to their + * default configuration. Providers are tried in descending priority order. + * + * @param sseParserConfig the SSE parser configuration to apply + * @return a new A2AHttpClient instance + * @throws IllegalStateException if no provider found or all providers failed to instantiate + */ + public static A2AHttpClient create(SSEParserConfig sseParserConfig) { + if (sseParserConfig == null) { + return create(); + } + return PROVIDERS.stream() + .flatMap(p -> { + try { + return Stream.of(p.create(sseParserConfig)); + } catch (Exception e) { + LOGGER.log(Level.WARNING, e, () -> "Provider " + p.name() + " skipped"); + return Stream.empty(); + } + }) + .findFirst() + .orElseThrow(() -> new IllegalStateException("No A2AHttpClientProvider could be instantiated")); + } + /** * Creates a new A2AHttpClient instance using a specific provider by name. * diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java index 9e8061360..a3fc774bc 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/A2AHttpClientProvider.java @@ -23,6 +23,19 @@ public interface A2AHttpClientProvider { */ A2AHttpClient create(); + /** + * Creates a new instance of an A2AHttpClient with the given {@link SSEParserConfig}. + * + *

Providers that support SSE parser configuration should override this method. + * The default implementation ignores {@code sseParserConfig} and delegates to {@link #create()}. + * + * @param sseParserConfig the SSE parser configuration to apply + * @return a new A2AHttpClient instance + */ + default A2AHttpClient create(SSEParserConfig sseParserConfig) { + return create(); + } + /** * Returns the priority of this provider. Higher priority providers are * tried first; the first one whose {@link #create()} succeeds is used. diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java index 8168b7f3b..a4629abec 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClient.java @@ -59,6 +59,7 @@ public class JdkA2AHttpClient implements A2AHttpClient { private final HttpClient httpClient; + private final SSEParserConfig sseParserConfig; private volatile @Nullable HttpClient noRedirectClient; /** @@ -81,7 +82,20 @@ public JdkA2AHttpClient() { this(HttpClient.newBuilder() .version(HttpClient.Version.HTTP_2) .followRedirects(HttpClient.Redirect.NEVER) - .build()); + .build(), SSEParserConfig.DEFAULT); + } + + /** + * Creates a new JDK-based HTTP client with secure defaults and custom SSE parser limits. + * + * @param sseParserConfig the SSE parser configuration to use for streaming responses + * @throws IllegalArgumentException if {@code sseParserConfig} is {@code null} + */ + public JdkA2AHttpClient(SSEParserConfig sseParserConfig) { + this(HttpClient.newBuilder() + .version(HttpClient.Version.HTTP_2) + .followRedirects(HttpClient.Redirect.NEVER) + .build(), sseParserConfig); } /** @@ -100,7 +114,20 @@ public JdkA2AHttpClient() { * @throws IllegalArgumentException if {@code httpClient} is {@code null} */ public JdkA2AHttpClient(HttpClient httpClient) { + this(httpClient, SSEParserConfig.DEFAULT); + } + + /** + * Creates a new JDK-based HTTP client using a caller-provided JDK {@link HttpClient} + * and custom SSE parser limits. + * + * @param httpClient the JDK HTTP client to delegate requests to + * @param sseParserConfig the SSE parser configuration to use for streaming responses + * @throws IllegalArgumentException if {@code httpClient} or {@code sseParserConfig} is {@code null} + */ + public JdkA2AHttpClient(HttpClient httpClient, SSEParserConfig sseParserConfig) { this.httpClient = checkNotNullParam("httpClient", httpClient); + this.sseParserConfig = checkNotNullParam("sseParserConfig", sseParserConfig); } @Override @@ -182,7 +209,7 @@ protected CompletableFuture asyncRequest( Consumer errorConsumer, Runnable completeRunnable ) { - ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer); + ServerSentEventParser sseParser = new ServerSentEventParser(messageConsumer, errorConsumer, sseParserConfig); AtomicBoolean useSseParser = new AtomicBoolean(false); AtomicBoolean errorNotified = new AtomicBoolean(false); StringBuilder nonSseBodyBuffer = new StringBuilder(); diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java index 4b53d41f6..0c3fb4992 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientProvider.java @@ -15,6 +15,14 @@ public A2AHttpClient create() { return new JdkA2AHttpClient(); } + /** + * {@inheritDoc} + */ + @Override + public A2AHttpClient create(SSEParserConfig sseParserConfig) { + return new JdkA2AHttpClient(sseParserConfig); + } + @Override public int priority() { return 0; // Lowest priority - fallback diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java new file mode 100644 index 000000000..75ee60287 --- /dev/null +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/SSEParserConfig.java @@ -0,0 +1,87 @@ +package org.a2aproject.sdk.client.http; + +/** + * Configuration for {@link ServerSentEventParser} limits. + * + *

All limits have safe defaults suitable for most deployments. Use the {@link Builder} + * to override individual values -- for example, to raise the per-event character budget + * for agents that return large artifacts. + * + *

Security note: setting {@code maxLineLength} to {@code 0} disables the + * per-line length check entirely, relying on {@code maxBufferChars} alone to bound memory. + * This means a single SSE line can consume up to {@code maxBufferChars} characters in memory. + * Only disable the per-line check when you trust the remote agent or have other safeguards + * (e.g. a reverse proxy that enforces its own line-length limit). + * + * @param maxLineLength max characters per raw SSE line (0 = disabled) + * @param maxBufferLines max {@code data:} lines per event block + * @param maxBufferChars max total characters across all {@code data:} values per event + */ +public record SSEParserConfig(int maxLineLength, int maxBufferLines, int maxBufferChars) { + + /** + * Default configuration: 1 MB per-line and per-event character limit, 1 000 data lines per event. + */ + public static final SSEParserConfig DEFAULT = new SSEParserConfig(1024 * 1024, 1000, 1024 * 1024); + + public SSEParserConfig { + if (maxLineLength < 0) { + throw new IllegalArgumentException("maxLineLength must be >= 0, got " + maxLineLength); + } + if (maxBufferLines <= 0) { + throw new IllegalArgumentException("maxBufferLines must be > 0, got " + maxBufferLines); + } + if (maxBufferChars <= 0) { + throw new IllegalArgumentException("maxBufferChars must be > 0, got " + maxBufferChars); + } + } + + /** + * Returns a new {@link Builder} initialized with the {@link #DEFAULT} values. + */ + public static Builder builder() { + return new Builder(); + } + + public static final class Builder { + private int maxLineLength = DEFAULT.maxLineLength; + private int maxBufferLines = DEFAULT.maxBufferLines; + private int maxBufferChars = DEFAULT.maxBufferChars; + + Builder() { + } + + /** + * Sets the maximum number of characters allowed in a single raw SSE line. + * Set to {@code 0} to disable the per-line check. + * + *

Security note: disabling this check removes the first line of + * defence against memory exhaustion from oversized SSE lines. Ensure + * {@code maxBufferChars} is set to an acceptable upper bound when disabling. + */ + public Builder maxLineLength(int maxLineLength) { + this.maxLineLength = maxLineLength; + return this; + } + + /** + * Sets the maximum number of {@code data:} lines allowed in a single event block. + */ + public Builder maxBufferLines(int maxBufferLines) { + this.maxBufferLines = maxBufferLines; + return this; + } + + /** + * Sets the maximum total characters across all {@code data:} values in a single event. + */ + public Builder maxBufferChars(int maxBufferChars) { + this.maxBufferChars = maxBufferChars; + return this; + } + + public SSEParserConfig build() { + return new SSEParserConfig(maxLineLength, maxBufferLines, maxBufferChars); + } + } +} diff --git a/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java b/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java index ea08fc97f..c9065f554 100644 --- a/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java +++ b/http-client/src/main/java/org/a2aproject/sdk/client/http/ServerSentEventParser.java @@ -15,9 +15,9 @@ public class ServerSentEventParser { private static final Logger LOGGER = Logger.getLogger(ServerSentEventParser.class.getName()); - private static final int MAX_BUFFER_SIZE = 1000; - private static final int MAX_BUFFER_CHARS = 1024 * 1024; // 1 MB (Java chars, so up to 2 MB in UTF-16; actual UTF-8 bytes may differ) - private static final int MAX_LINE_LENGTH = 65536; // 64 KB + private final int maxLineLength; + private final int maxBufferLines; + private final int maxBufferChars; private final Consumer eventConsumer; private final @Nullable Consumer errorConsumer; @@ -35,12 +35,20 @@ public class ServerSentEventParser { private boolean skippingCurrentEvent = false; public ServerSentEventParser(Consumer eventConsumer) { - this(eventConsumer, null); + this(eventConsumer, null, SSEParserConfig.DEFAULT); } public ServerSentEventParser(Consumer eventConsumer, @Nullable Consumer errorConsumer) { + this(eventConsumer, errorConsumer, SSEParserConfig.DEFAULT); + } + + public ServerSentEventParser(Consumer eventConsumer, @Nullable Consumer errorConsumer, + SSEParserConfig config) { this.eventConsumer = eventConsumer; this.errorConsumer = errorConsumer; + this.maxLineLength = config.maxLineLength(); + this.maxBufferLines = config.maxBufferLines(); + this.maxBufferChars = config.maxBufferChars(); } /** @@ -54,8 +62,8 @@ public void processLine(@Nullable String line) { } // Check line length to prevent DoS; corrupt the current event so it is not dispatched - if (line.length() > MAX_LINE_LENGTH) { - handleError(new IllegalArgumentException("Line exceeds maximum length of " + MAX_LINE_LENGTH + " characters")); + if (maxLineLength > 0 && line.length() > maxLineLength) { + handleError(new IllegalArgumentException("Line exceeds maximum length of " + maxLineLength + " characters")); skippingCurrentEvent = true; dataBuffer.clear(); dataBufferChars = 0; @@ -99,16 +107,16 @@ private void processField(String field, String value) { switch (field) { case "data" -> { // Check line count to prevent DoS; corrupt and skip the rest of this event block - if (dataBuffer.size() >= MAX_BUFFER_SIZE) { - handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + MAX_BUFFER_SIZE + " lines")); + if (dataBuffer.size() >= maxBufferLines) { + handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + maxBufferLines + " lines")); skippingCurrentEvent = true; dataBuffer.clear(); dataBufferChars = 0; return; } // Check total char count to prevent OOM on large streams - if (dataBufferChars + value.length() > MAX_BUFFER_CHARS) { - handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + MAX_BUFFER_CHARS + " chars")); + if (dataBufferChars + value.length() > maxBufferChars) { + handleError(new IllegalStateException("SSE data buffer exceeded maximum size of " + maxBufferChars + " chars")); skippingCurrentEvent = true; dataBuffer.clear(); dataBufferChars = 0; @@ -146,9 +154,14 @@ private void processField(String field, String value) { } private void dispatchEvent() { - // Per SSE spec: update lastEventId before checking data, so ID-only events (e.g. heartbeats) are tracked - if (currentEventId != null) { + // Per SSE spec §9.2.6: copy currentEventId → lastEventId at dispatch, but only for blocks that + // were not skipped. A corrupt/oversized block must not advance the reconnect cursor even if its + // id: field was parsed before the violation was detected. + if (!skippingCurrentEvent && currentEventId != null) { lastEventId = currentEventId; + } else if (skippingCurrentEvent) { + // Roll back the event ID buffer so the skipped block's id: cannot leak into subsequent events. + currentEventId = lastEventId; } String data = String.join("\n", dataBuffer); diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java index 814b01c33..02025b816 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientFactoryTest.java @@ -56,11 +56,27 @@ public void testCreateWithInvalidProviderNameThrows() { public void testCreateWithNullProviderNameThrows() { assertThrows( IllegalArgumentException.class, - () -> A2AHttpClientFactory.create(null), + () -> A2AHttpClientFactory.create((String) null), "Factory should throw IllegalArgumentException for null provider name" ); } + @Test + public void testCreateWithNullSseParserConfigDelegatesToCreate() { + A2AHttpClient client = A2AHttpClientFactory.create((SSEParserConfig) null); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client); + } + + @Test + public void testCreateWithSseParserConfigReturnsJdkClient() { + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build(); + A2AHttpClient client = A2AHttpClientFactory.create(config); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client, + "Factory should return JdkA2AHttpClient with custom SSEParserConfig"); + } + @Test public void testCreateWithEmptyProviderNameThrows() { assertThrows( diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java index 525b0b505..5d9543e47 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/A2AHttpClientProviderTest.java @@ -14,6 +14,15 @@ public void testJdkProviderCreatesClient() { assertInstanceOf(JdkA2AHttpClient.class, client); } + @Test + public void testJdkProviderCreatesClientWithSseParserConfig() { + JdkA2AHttpClientProvider provider = new JdkA2AHttpClientProvider(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(4 * 1024 * 1024).build(); + A2AHttpClient client = provider.create(config); + assertNotNull(client); + assertInstanceOf(JdkA2AHttpClient.class, client); + } + @Test public void testJdkProviderPriority() { JdkA2AHttpClientProvider provider = new JdkA2AHttpClientProvider(); diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java index 63644fd1b..690277148 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/AbstractA2AHttpClientSSETest.java @@ -5,10 +5,12 @@ import static org.junit.jupiter.api.Assertions.assertNotNull; import static org.junit.jupiter.api.Assertions.assertNull; import static org.junit.jupiter.api.Assertions.assertTrue; +import static org.junit.jupiter.api.Assumptions.assumeTrue; import static org.mockserver.model.HttpRequest.request; import static org.mockserver.model.HttpResponse.response; import org.a2aproject.sdk.common.A2AErrorMessages; +import org.jspecify.annotations.Nullable; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -29,6 +31,15 @@ public abstract class AbstractA2AHttpClientSSETest { protected abstract A2AHttpClient createClient(); + /** + * Creates a client with a custom {@link SSEParserConfig}. + * Returns {@code null} if the implementation does not support SSEParserConfig, + * in which case the SSE config integration tests are skipped. + */ + protected @Nullable A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return null; + } + @BeforeEach public void setup() { mockServer = ClientAndServer.startClientAndServer(0); @@ -377,4 +388,72 @@ public void testPostSSETypedEvents() throws Exception { assertEquals("99", events.get(0).id()); assertEquals("done", events.get(0).data()); } + + @Test + public void testCustomSSEParserConfigRejectsOversizedLine() throws Exception { + A2AHttpClient configClient = createClient(SSEParserConfig.builder().maxLineLength(50).build()); + assumeTrue(configClient != null, "Implementation does not support SSEParserConfig"); + + // 60-char payload exceeds the 50-char per-line limit + String oversizedPayload = "x".repeat(60); + mockServer + .when(request().withMethod("POST").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: " + oversizedPayload + "\n\n")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + configClient.createPost() + .url(getBaseUrl() + "/sse") + .body("{}") + .postAsyncSSE( + events::add, + e -> { + error.set(e); + latch.countDown(); + }, + latch::countDown + ); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNotNull(error.get(), "Custom maxLineLength should reject oversized SSE line"); + assertEquals(0, events.size(), "Oversized event must not be dispatched"); + } + + @Test + public void testCustomSSEParserConfigAcceptsLineWithinLimit() throws Exception { + A2AHttpClient configClient = createClient(SSEParserConfig.builder().maxLineLength(200).build()); + assumeTrue(configClient != null, "Implementation does not support SSEParserConfig"); + + // 100-char payload is within the 200-char per-line limit + String payload = "x".repeat(100); + mockServer + .when(request().withMethod("POST").withPath("/sse")) + .respond(response() + .withStatusCode(200) + .withHeader("Content-Type", "text/event-stream") + .withBody("data: " + payload + "\n\n")); + + CountDownLatch latch = new CountDownLatch(1); + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + + configClient.createPost() + .url(getBaseUrl() + "/sse") + .body("{}") + .postAsyncSSE( + events::add, + error::set, + latch::countDown + ); + + assertTrue(latch.await(5, TimeUnit.SECONDS)); + assertNull(error.get(), "Payload within limit should not trigger an error"); + assertEquals(1, events.size()); + assertEquals(payload, events.get(0).data()); + } } diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java new file mode 100644 index 000000000..695bfe4aa --- /dev/null +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientSSETest.java @@ -0,0 +1,14 @@ +package org.a2aproject.sdk.client.http; + +public class JdkA2AHttpClientSSETest extends AbstractA2AHttpClientSSETest { + + @Override + protected A2AHttpClient createClient() { + return new JdkA2AHttpClient(); + } + + @Override + protected A2AHttpClient createClient(SSEParserConfig sseParserConfig) { + return new JdkA2AHttpClient(sseParserConfig); + } +} diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java index 0a8780119..a517a1ba1 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/JdkA2AHttpClientTest.java @@ -94,7 +94,7 @@ public void testConstructorUsesProvidedHttpClient() throws Exception { @Test public void testConstructorRejectsNullHttpClient() { - assertThrows(IllegalArgumentException.class, () -> new JdkA2AHttpClient(null), "foo"); + assertThrows(IllegalArgumentException.class, () -> new JdkA2AHttpClient((java.net.http.HttpClient) null), "foo"); } @Test diff --git a/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java b/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java index 0492d1152..5774e4b77 100644 --- a/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java +++ b/http-client/src/test/java/org/a2aproject/sdk/client/http/ServerSentEventParserTest.java @@ -2,8 +2,10 @@ import static org.junit.jupiter.api.Assertions.assertDoesNotThrow; import static org.junit.jupiter.api.Assertions.assertEquals; +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.assertThrows; import java.util.ArrayList; import java.util.List; @@ -370,11 +372,12 @@ public void testErrorConsumerCalledForNullLine() { public void testErrorConsumerCalledForLineTooLong() { List events = new ArrayList<>(); AtomicReference error = new AtomicReference<>(); - ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set); + SSEParserConfig config = SSEParserConfig.builder().maxLineLength(1000).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); // Oversized line mid-event: the whole event block is discarded parser.processLine("data: before overflow"); - String longLine = "data: " + "x".repeat(65537); + String longLine = "data: " + "x".repeat(1001); parser.processLine(longLine); // Subsequent lines in the same block are skipped parser.processLine("data: should be skipped"); @@ -425,8 +428,7 @@ public void testErrorConsumerCalledForBufferByteOverflow() { AtomicReference error = new AtomicReference<>(); ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set); - // Value is 65530 chars so the full line ("data: " + value = 65536) stays within the per-line - // limit; 17 such lines (17 * 65530 = 1,114,010 bytes) exceed the 1MB buffer byte limit. + // Each value is 65530 chars; 17 such lines (17 * 65530 = 1,114,010 chars) exceed the 1 MB buffer char limit. String bigValue = "x".repeat(65530); for (int i = 0; i < 17; i++) { parser.processLine("data: " + bigValue); @@ -510,4 +512,327 @@ public void testCRLFLineTerminatorsPreservedInValue() { assertEquals(1, events.size()); assertEquals("value\r", events.get(0).data()); } + + @Test + public void testLargeJsonRpcResponseRejectedByOldLimit() { + // Reproducer: a 70 KB payload exceeds the old 64 KB per-line limit but fits within the + // new 1 MB default, proving the limit raise fixes real-world large JSON-RPC responses. + String hugeJson = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"" + "x".repeat(70_000) + "\"}"; + + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig oldLimit = SSEParserConfig.builder().maxLineLength(65536).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, oldLimit); + + parser.processLine("data: " + hugeJson); + parser.processLine(""); + + assertEquals(0, events.size(), "Event must be rejected under the old 64 KB limit"); + assertEquals(1, errors.size()); + assertInstanceOf(IllegalArgumentException.class, errors.get(0)); + + // Same payload split across two data: lines parses fine under the old limit + events.clear(); + errors.clear(); + String half1 = hugeJson.substring(0, hugeJson.length() / 2); + String half2 = hugeJson.substring(hugeJson.length() / 2); + parser.processLine("data: " + half1); + parser.processLine("data: " + half2); + parser.processLine(""); + + assertEquals(0, errors.size(), "Split payload should not trigger any error"); + assertEquals(1, events.size()); + assertEquals(half1 + "\n" + half2, events.get(0).data()); + } + + @Test + public void testLargeJsonRpcResponseAcceptedByDefault() { + // With the raised 1 MB default the same payload parses on a single line + String hugeJson = "{\"jsonrpc\":\"2.0\",\"id\":1,\"result\":\"" + "x".repeat(70_000) + "\"}"; + + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add); + + parser.processLine("data: " + hugeJson); + parser.processLine(""); + + assertEquals(0, errors.size(), "70 KB line should be accepted with default 1 MB limit"); + assertEquals(1, events.size()); + assertEquals(hugeJson, events.get(0).data()); + } + + @Test + public void testLargeSingleLineEventAcceptedByDefault() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set); + + // 200 KB single data: line -- previously rejected at 64 KB, now accepted with 1 MB default + String largeJson = "{\"result\":\"" + "x".repeat(200_000) + "\"}"; + parser.processLine("data: " + largeJson); + parser.processLine(""); + + assertNull(error.get(), "200 KB line should be accepted with default 1 MB limit"); + assertEquals(1, events.size()); + assertEquals(largeJson, events.get(0).data()); + } + + @Test + public void testDisabledLineLengthCheck() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxLineLength(0) + .maxBufferChars(2 * 1024 * 1024) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + // With maxLineLength=0 (disabled), even very large lines are accepted + String hugeLine = "data: " + "x".repeat(1_500_000); + parser.processLine(hugeLine); + parser.processLine(""); + + assertNull(error.get(), "Line length check should be disabled when maxLineLength=0"); + assertEquals(1, events.size()); + assertEquals("x".repeat(1_500_000), events.get(0).data()); + } + + @Test + public void testCustomBufferLineLimit() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxBufferLines(5) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + for (int i = 0; i < 5; i++) { + parser.processLine("data: line" + i); + } + assertNull(error.get(), "No error expected at exactly the limit"); + + parser.processLine("data: overflow"); + assertNotNull(error.get(), "errorConsumer should be called when custom buffer line limit exceeded"); + parser.processLine(""); + assertEquals(0, events.size(), "Corrupted event block must not be dispatched"); + } + + @Test + public void testCustomBufferCharLimit() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxBufferChars(100) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + parser.processLine("data: " + "x".repeat(101)); + assertNotNull(error.get(), "errorConsumer should be called when custom buffer char limit exceeded"); + parser.processLine(""); + assertEquals(0, events.size(), "Corrupted event block must not be dispatched"); + } + + @Test + public void testParserRecoveryAfterCustomLimitViolation() { + List events = new ArrayList<>(); + AtomicReference error = new AtomicReference<>(); + SSEParserConfig config = SSEParserConfig.builder() + .maxBufferLines(2) + .build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, error::set, config); + + parser.processLine("data: line0"); + parser.processLine("data: line1"); + parser.processLine("data: overflow"); + assertNotNull(error.get(), "errorConsumer should be called when custom buffer line limit exceeded"); + parser.processLine(""); + assertEquals(0, events.size(), "Corrupted event block must not be dispatched"); + + error.set(null); + parser.processLine("data: ok"); + parser.processLine(""); + assertNull(error.get(), "No error expected after recovery"); + assertEquals(1, events.size(), "Parser should recover after custom limit violation"); + } + + @Test + public void testSSEParserConfigDefaults() { + SSEParserConfig config = SSEParserConfig.DEFAULT; + assertEquals(1024 * 1024, config.maxLineLength()); + assertEquals(1000, config.maxBufferLines()); + assertEquals(1024 * 1024, config.maxBufferChars()); + } + + @Test + public void testSSEParserConfigBuilder() { + SSEParserConfig config = SSEParserConfig.builder() + .maxLineLength(500_000) + .maxBufferLines(2000) + .maxBufferChars(4 * 1024 * 1024) + .build(); + assertEquals(500_000, config.maxLineLength()); + assertEquals(2000, config.maxBufferLines()); + assertEquals(4 * 1024 * 1024, config.maxBufferChars()); + } + + @Test + public void testSSEParserConfigValidation() { + assertDoesNotThrow(() -> SSEParserConfig.builder().maxLineLength(0).build(), + "maxLineLength=0 (disabled) should be allowed"); + + assertThrows(IllegalArgumentException.class, + () -> SSEParserConfig.builder().maxLineLength(-1).build()); + + assertThrows(IllegalArgumentException.class, + () -> SSEParserConfig.builder().maxBufferLines(0).build()); + + assertThrows(IllegalArgumentException.class, + () -> SSEParserConfig.builder().maxBufferChars(0).build()); + } + + // --- lastEventId / skipping interaction --- + + @Test + public void testLastEventIdNotAdvancedWhenBlockSkippedByLineTooLong() { + // Regression: id: in a corrupt block must not update the reconnect cursor. + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + // Limit long enough for "id: good-id" (11) and "data: ok" (8), but shorter than the oversized data line. + SSEParserConfig config = SSEParserConfig.builder().maxLineLength(50).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + // Good event that sets lastEventId to "good-id" + parser.processLine("id: good-id"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals(1, events.size(), "Good event should be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId should be set by good event"); + + // Corrupt block: id: comes before the oversized line + parser.processLine("id: bad-id"); + parser.processLine("data: " + "x".repeat(51)); // triggers skip + parser.processLine(""); // end of corrupt block + assertEquals(1, events.size(), "Corrupt block must not be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId must not advance for a skipped block"); + } + + @Test + public void testLastEventIdNotAdvancedWhenBlockSkippedByBufferLineOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferLines(2).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good-id"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good-id", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad-id"); + parser.processLine("data: line0"); + parser.processLine("data: line1"); + parser.processLine("data: overflow"); // triggers skip + parser.processLine(""); + assertEquals(1, events.size(), "Only the first good event should be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId must not advance for a skipped block"); + } + + @Test + public void testSkippedBlockIdDoesNotPoisonNextEventByLineTooLong() { + // Regression: after a skipped block, currentEventId must be rolled back so a subsequent + // event without an id: field does not inherit the skipped block's id. + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxLineLength(50).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + // Good event + parser.processLine("id: good"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good", parser.getLastEventId(), "lastEventId should be set by good event"); + + // Corrupt block with a different id + parser.processLine("id: bad"); + parser.processLine("data: " + "x".repeat(51)); + parser.processLine(""); + + // Next valid event has no id: field + parser.processLine("data: next-valid-event"); + parser.processLine(""); + + assertEquals(2, events.size()); + assertEquals("good", events.get(1).id(), "Next event must carry the pre-skip id, not the skipped block's id"); + assertEquals("good", parser.getLastEventId(), "lastEventId must not be poisoned by the skipped block"); + } + + @Test + public void testSkippedBlockIdDoesNotPoisonNextEventByBufferOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferLines(2).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad"); + parser.processLine("data: line0"); + parser.processLine("data: line1"); + parser.processLine("data: overflow"); + parser.processLine(""); + + parser.processLine("data: next-valid-event"); + parser.processLine(""); + + assertEquals(2, events.size()); + assertEquals("good", events.get(1).id(), "Next event must carry the pre-skip id, not the skipped block's id"); + assertEquals("good", parser.getLastEventId(), "lastEventId must not be poisoned by the skipped block"); + } + + @Test + public void testSkippedBlockIdDoesNotPoisonNextEventByCharOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(20).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad"); + parser.processLine("data: " + "x".repeat(21)); + parser.processLine(""); + + parser.processLine("data: next-valid-event"); + parser.processLine(""); + + assertEquals(2, events.size()); + assertEquals("good", events.get(1).id(), "Next event must carry the pre-skip id, not the skipped block's id"); + assertEquals("good", parser.getLastEventId(), "lastEventId must not be poisoned by the skipped block"); + } + + @Test + public void testLastEventIdNotAdvancedWhenBlockSkippedByBufferCharOverflow() { + List events = new ArrayList<>(); + List errors = new ArrayList<>(); + SSEParserConfig config = SSEParserConfig.builder().maxBufferChars(20).build(); + ServerSentEventParser parser = new ServerSentEventParser(events::add, errors::add, config); + + parser.processLine("id: good-id"); + parser.processLine("data: ok"); + parser.processLine(""); + assertEquals("good-id", parser.getLastEventId(), "lastEventId should be set by good event"); + + parser.processLine("id: bad-id"); + parser.processLine("data: " + "x".repeat(21)); // triggers skip + parser.processLine(""); + assertEquals(1, events.size(), "Only the first good event should be dispatched"); + assertEquals("good-id", parser.getLastEventId(), "lastEventId must not advance for a skipped block"); + } }