You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@camel.apache.org by ja...@apache.org on 2026/06/19 15:02:21 UTC
(camel) branch main updated: CAMEL-23804: Fix SSE streaming on camel-platform-http-vertx
This is an automated email from the ASF dual-hosted git repository.
jamesnetherton pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/camel.git
The following commit(s) were added to refs/heads/main by this push:
new 32bb460aa00e CAMEL-23804: Fix SSE streaming on camel-platform-http-vertx
32bb460aa00e is described below
commit 32bb460aa00ef206d18b51d67bf649244820b473
Author: James Netherton <ja...@users.noreply.github.com>
AuthorDate: Fri Jun 19 16:02:14 2026 +0100
CAMEL-23804: Fix SSE streaming on camel-platform-http-vertx
AsyncInputStream greedily filled its 16KB buffer before emitting data,
which blocked SSE events behind the next queue poll. Add an eagerFlush
mode that emits each chunk immediately, and activate it automatically
for text/event-stream responses. Also slice the output buffer to the
actual bytes written to prevent trailing zeroes on partial reads.
Update camel-a2a docs to remove the Vert.x SSE streaming limitation
now that platform-http delivers events eagerly.
Co-authored-by: Claude Opus 4.6 (1M context) <no...@anthropic.com>
---
.../camel-a2a/src/main/docs/a2a-component.adoc | 22 +---
.../platform/http/vertx/AsyncInputStream.java | 18 ++-
.../http/vertx/VertxPlatformHttpSupport.java | 6 +-
.../http/vertx/VertxPlatformHttpSseTest.java | 138 +++++++++++++++++++++
4 files changed, 162 insertions(+), 22 deletions(-)
diff --git a/components/camel-ai/camel-a2a/src/main/docs/a2a-component.adoc b/components/camel-ai/camel-a2a/src/main/docs/a2a-component.adoc
index 03c08d7ca30f..4515f9949323 100644
--- a/components/camel-ai/camel-a2a/src/main/docs/a2a-component.adoc
+++ b/components/camel-ai/camel-a2a/src/main/docs/a2a-component.adoc
@@ -36,7 +36,7 @@ Current Preview limitations:
* Extended Agent Card / `GetExtendedAgentCard` is not implemented in this first release.
* Consumers validate operation requests by default. Local-only examples in this page set `validateAuth=false`; network-exposed agents should use an agent card with `securitySchemes` and `securityRequirements` plus `apiKey`, `bearerToken`, or `oauthProfile` endpoint configuration.
* The REST binding uses A2A custom-method paths containing colons, such as `/message:send`. These paths collide on Vert.x/platform-http. Use `protocolBinding=JSONRPC` with platform-http, or use `httpServerComponent=undertow` or `httpServerComponent=jetty` for REST custom-method routes.
-* Real-time SSE streaming is verified with Undertow and Jetty. Vert.x/platform-http buffers `InputStream` responses and does not deliver events as they are emitted.
+* Real-time SSE streaming is verified with Vert.x/platform-http, Undertow, and Jetty.
* The default task store is in-memory and single-JVM. Register a custom `A2ATaskStore` for durable or cross-node task state.
Maven users will need to add the following dependency to their `pom.xml`:
@@ -51,7 +51,7 @@ Maven users will need to add the following dependency to their `pom.xml`:
</dependency>
----
-Consumer routes also need an HTTP server component at runtime. For example, add `camel-platform-http-vertx` for JSON-RPC agents, or `camel-undertow` / `camel-jetty` for REST custom-method routes and real-time SSE streaming. See <<Dependencies>> and <<HTTP Server Component Discovery>>.
+Consumer routes also need an HTTP server component at runtime. For example, add `camel-platform-http-vertx` for JSON-RPC agents (including SSE streaming), or `camel-undertow` / `camel-jetty` for REST custom-method routes. See <<Dependencies>> and <<HTTP Server Component Discovery>>.
== URI Format
@@ -249,7 +249,7 @@ The A2A REST protocol uses Google's Custom Method convention with colons in path
This means operation serving is not reliable with REST binding on Vert.x/platform-http. For example, `POST /message:stream` can be handled by the `SendMessage` route instead of the streaming route.
-*Workaround:* Use `protocolBinding=JSONRPC` for agents running on Vert.x (the default `platform-http`). Alternatively, use `httpServerComponent=jetty` or `httpServerComponent=undertow`; their routers treat colons as literal characters. For real-time SSE streaming, use Jetty or Undertow because Vert.x/platform-http buffers `InputStream` responses.
+*Workaround:* Use `protocolBinding=JSONRPC` for agents running on Vert.x (the default `platform-http`). Alternatively, use `httpServerComponent=jetty` or `httpServerComponent=undertow`; their routers treat colons as literal characters.
=== Authentication
@@ -559,19 +559,11 @@ The consumer supports two SSE streaming patterns:
* *`SendStreamingMessage`* — The stream starts with the submitted `Task`, followed by progress events from `A2AProgress` or `${a2a:emit()}`. When the route completes, the body is emitted as the final message.
* *`SubscribeToTask`* — Clients subscribe via `POST /tasks/\{id}:subscribe` to receive real-time SSE updates for an existing non-terminal task. The component uses an Event-to-Stream Bridge (`QueueStreamEmitter` + `SseQueueInputStream`) with heartbeat comments to keep the connection alive. The stream sends the current `Task` first and ends when the task later reaches a terminal state. Subscribing to an already-terminal task returns `UnsupportedOperationError`.
-[IMPORTANT]
+[NOTE]
====
*HTTP Server Component for Streaming*
-For real-time SSE streaming, use `httpServerComponent=undertow` or `httpServerComponent=jetty`. The default Vert.x `platform-http` buffers `InputStream` responses due to an `AsyncInputStream` greedy-fill loop, causing all events to arrive at once.
-
-[source,yaml]
-----
-parameters:
- httpServerComponent: undertow
-----
-
-This requires `camel-undertow` (or `camel-jetty`) on the classpath.
+Real-time SSE streaming works with all tested HTTP server components: Vert.x/platform-http, Undertow, and Jetty.
====
SSE streaming parameters:
@@ -1654,11 +1646,9 @@ The preview component does not auto-create arbitrary HTTP server components and
Tested behavior in this Preview release:
-* `camel-platform-http` / Vert.x - suitable for JSON-RPC agents and serving the public agent card. REST custom-method paths containing colons can collide, and SSE `InputStream` responses are buffered.
+* `camel-platform-http` / Vert.x - suitable for JSON-RPC agents (including SSE streaming) and serving the public agent card. REST custom-method paths containing colons can collide.
* `camel-undertow` and `camel-jetty` - suitable for REST custom-method paths and real-time SSE streaming.
-For SSE streaming, specify `httpServerComponent=undertow` or `httpServerComponent=jetty` and add the matching component dependency.
-
== Data Format
The `dataFormat` parameter controls what the exchange body contains, following the same convention as the xref:cxf-component.adoc[CXF component]. It applies to both consumer (inbound request) and producer (response) sides.
diff --git a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java
index b2ea643b08d2..17825ef417d3 100644
--- a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java
+++ b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/AsyncInputStream.java
@@ -48,6 +48,7 @@ public class AsyncInputStream implements ReadStream<Buffer> {
private final Vertx vertx;
private final Context context;
private final InboundBuffer<Buffer> queue;
+ private final boolean eagerFlush;
private long readPos;
private boolean closed;
private boolean readInProgress;
@@ -56,9 +57,14 @@ public class AsyncInputStream implements ReadStream<Buffer> {
private Handler<Throwable> exceptionHandler;
public AsyncInputStream(Vertx vertx, Context context, InputStream inputStream) {
+ this(vertx, context, inputStream, false);
+ }
+
+ public AsyncInputStream(Vertx vertx, Context context, InputStream inputStream, boolean eagerFlush) {
this.vertx = vertx;
this.context = context;
this.channel = Channels.newChannel(inputStream);
+ this.eagerFlush = eagerFlush;
this.queue = new InboundBuffer<>(context, 0);
queue.handler(buffer -> {
if (buffer.length() > 0) {
@@ -222,22 +228,24 @@ public class AsyncInputStream implements ReadStream<Buffer> {
// EOF
context.runOnContext((v) -> {
buffer.flip();
+ int written = offset + buffer.limit();
writeBuff.setBytes(offset, buffer);
buffer.compact();
- handler.handle(Future.succeededFuture(writeBuff));
+ handler.handle(Future.succeededFuture(writeBuff.slice(0, written)));
});
- } else if (buffer.hasRemaining()) {
- // Read from the next offset
+ } else if (!eagerFlush && buffer.hasRemaining()) {
+ // Read from the next offset to fill the buffer
context.runOnContext((v) -> {
doRead(writeBuff, offset, buffer, position + bytesRead, handler);
});
} else {
- // All data is written
+ // Emit what we have (always for eagerFlush, or when buffer is full)
context.runOnContext((v) -> {
buffer.flip();
+ int written = offset + buffer.limit();
writeBuff.setBytes(offset, buffer);
buffer.compact();
- handler.handle(Future.succeededFuture(writeBuff));
+ handler.handle(Future.succeededFuture(writeBuff.slice(0, written)));
});
}
} else {
diff --git a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java
index 3c9eba48653b..b731f93fadd4 100644
--- a/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java
+++ b/components/camel-platform-http-vertx/src/main/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSupport.java
@@ -237,8 +237,12 @@ public final class VertxPlatformHttpSupport {
Vertx vertx = ctx.vertx();
Context context = vertx.getOrCreateContext();
+ // For SSE streams, flush each chunk immediately instead of greedily filling the buffer
+ String contentType = response.headers().get("Content-Type");
+ boolean eagerFlush = contentType != null && contentType.startsWith("text/event-stream");
+
// Process the InputStream async to avoid blocking the Vert.x event loop on large responses
- AsyncInputStream asyncInputStream = new AsyncInputStream(vertx, context, is);
+ AsyncInputStream asyncInputStream = new AsyncInputStream(vertx, context, is, eagerFlush);
asyncInputStream.exceptionHandler(promise::fail);
asyncInputStream.endHandler(event -> endHandler(promise, response, asyncInputStream));
diff --git a/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSseTest.java b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSseTest.java
new file mode 100644
index 000000000000..52ca8e7f147b
--- /dev/null
+++ b/components/camel-platform-http-vertx/src/test/java/org/apache/camel/component/platform/http/vertx/VertxPlatformHttpSseTest.java
@@ -0,0 +1,138 @@
+/*
+ * Licensed to the Apache Software Foundation (ASF) under one or more
+ * contributor license agreements. See the NOTICE file distributed with
+ * this work for additional information regarding copyright ownership.
+ * The ASF licenses this file to You under the Apache License, Version 2.0
+ * (the "License"); you may not use this file except in compliance with
+ * the License. You may obtain a copy of the License at
+ *
+ * http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.apache.camel.component.platform.http.vertx;
+
+import java.io.PipedInputStream;
+import java.io.PipedOutputStream;
+import java.nio.charset.StandardCharsets;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.TimeUnit;
+import java.util.concurrent.atomic.AtomicReference;
+
+import io.vertx.core.http.HttpClient;
+import io.vertx.core.http.HttpMethod;
+import org.apache.camel.CamelContext;
+import org.apache.camel.Exchange;
+import org.apache.camel.builder.RouteBuilder;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.assertEquals;
+import static org.junit.jupiter.api.Assertions.assertTrue;
+
+public class VertxPlatformHttpSseTest {
+
+ @Test
+ void testSseEventsDeliveredEagerly() throws Exception {
+ final CamelContext context = VertxPlatformHttpEngineTest.createCamelContext();
+
+ // Writer waits on this latch after sending event-1.
+ // If event-1 is flushed eagerly, the client reads it and counts down,
+ // unblocking the writer to send event-2.
+ // If events are batched, the client never sees event-1 until the stream
+ // closes, so the latch times out and the test fails.
+ CountDownLatch event1Received = new CountDownLatch(1);
+ AtomicReference<Throwable> writerError = new AtomicReference<>();
+
+ HttpClient client = null;
+ try {
+ context.addRoutes(new RouteBuilder() {
+ @Override
+ public void configure() {
+ from("platform-http:/sse")
+ .process(exchange -> {
+ PipedOutputStream out = new PipedOutputStream();
+ PipedInputStream in = new PipedInputStream(out);
+
+ exchange.getMessage().setHeader(Exchange.CONTENT_TYPE, "text/event-stream");
+ exchange.getMessage().setBody(in);
+
+ Thread writer = new Thread(() -> {
+ try {
+ out.write("data: event-1\n\n".getBytes(StandardCharsets.UTF_8));
+ out.flush();
+
+ assertTrue(event1Received.await(5, TimeUnit.SECONDS),
+ "Client did not receive event-1 before timeout — events may be buffered");
+
+ out.write("data: event-2\n\n".getBytes(StandardCharsets.UTF_8));
+ out.flush();
+ out.close();
+ } catch (Throwable e) {
+ writerError.set(e);
+ try {
+ out.close();
+ } catch (Exception ignored) {
+ }
+ }
+ });
+ writer.setDaemon(true);
+ writer.start();
+ });
+ }
+ });
+
+ VertxPlatformHttpEngineTest.startCamelContext(context);
+
+ VertxPlatformHttpServer server = context.hasService(VertxPlatformHttpServer.class);
+ client = server.getVertx().createHttpClient();
+
+ StringBuilder received = new StringBuilder();
+ CompletableFuture<Void> streamDone = new CompletableFuture<>();
+
+ client.request(HttpMethod.GET, server.getPort(), "localhost", "/sse")
+ .onSuccess(request -> {
+ request.response()
+ .onSuccess(response -> {
+ assertEquals(200, response.statusCode());
+ assertEquals("text/event-stream", response.getHeader("Content-Type"));
+
+ response.handler(buffer -> {
+ String chunk = buffer.toString(StandardCharsets.UTF_8);
+ received.append(chunk);
+
+ if (chunk.contains("event-1")) {
+ event1Received.countDown();
+ }
+ });
+ response.endHandler(v -> streamDone.complete(null));
+ response.exceptionHandler(streamDone::completeExceptionally);
+ })
+ .onFailure(streamDone::completeExceptionally);
+ request.end();
+ })
+ .onFailure(streamDone::completeExceptionally);
+
+ streamDone.get(10, TimeUnit.SECONDS);
+
+ String[] events = received.toString().split("\n\n");
+ assertEquals(2, events.length);
+ assertEquals("data: event-1", events[0]);
+ assertEquals("data: event-2", events[1]);
+
+ Throwable error = writerError.get();
+ if (error != null) {
+ throw new AssertionError("Writer thread failed", error);
+ }
+ } finally {
+ if (client != null) {
+ client.close().toCompletionStage().toCompletableFuture().get(5, TimeUnit.SECONDS);
+ }
+ context.stop();
+ }
+ }
+}