Merge branch 'main' into issue_765

Author: ehsavoie

reference/jsonrpc/src/main/java/io/a2a/server/apps/quarkus/A2AServerRoutes.java

@@ -299,15 +299,15 @@ public void invokeJSONRPCHandler(@Body String body, RoutingContext rc) {
                         .putHeader(CONTENT_TYPE, APPLICATION_JSON)
                         .end(serializeResponse(error));
             } else if (streaming) {
-                final Multi<? extends A2AResponse<?>> finalStreamingResponse = streamingResponse;
-                executor.execute(() -> {
-                    // Convert Multi<A2AResponse> to Multi<String> with SSE formatting
-                    AtomicLong eventIdCounter = new AtomicLong(0);
-                    Multi<String> sseEvents = finalStreamingResponse
-                            .map(response -> SseFormatter.formatResponseAsSSE(response, eventIdCounter.getAndIncrement()));
-                    // Write SSE-formatted strings to HTTP response
-                    MultiSseSupport.writeSseStrings(sseEvents, rc, context);
-                });
+                // Convert Multi<A2AResponse> to Multi<String> with SSE formatting
+                // CRITICAL: Subscribe synchronously to avoid race condition where EventConsumer
+                // starts emitting events before MultiSseSupport subscribes. The executor.execute()
+                // wrapper caused 100-600ms delays before subscription, causing events to be lost.
+                AtomicLong eventIdCounter = new AtomicLong(0);
+                Multi<String> sseEvents = streamingResponse
+                        .map(response -> SseFormatter.formatResponseAsSSE(response, eventIdCounter.getAndIncrement()));
+                // Write SSE-formatted strings to HTTP response
+                MultiSseSupport.writeSseStrings(sseEvents, rc, context);
 
             } else {
                 rc.response()
@@ -783,7 +783,17 @@ public void onNext(String sseEvent) {
                         if (headers.get(CONTENT_TYPE) == null) {
                             headers.set(CONTENT_TYPE, SERVER_SENT_EVENTS);
                         }
+                        // Additional SSE headers to prevent buffering
+                        headers.set("Cache-Control", "no-cache");
+                        headers.set("X-Accel-Buffering", "no");  // Disable nginx buffering
                         response.setChunked(true);
+
+                        // CRITICAL: Disable write queue max size to prevent buffering
+                        // Vert.x buffers writes by default - we need immediate flushing for SSE
+                        response.setWriteQueueMaxSize(1);
+
+                        // Send initial SSE comment to kickstart the stream
+                        response.write(": SSE stream started\n\n");
                     }
 
                     // Write SSE-formatted string to response

reference/rest/src/main/java/io/a2a/server/rest/quarkus/A2AServerRoutes.java

@@ -221,15 +221,15 @@ public void sendMessageStreaming(@Body String body, RoutingContext rc) {
             if (error != null) {
                 sendResponse(rc, error);
             } else if (streamingResponse != null) {
-                final HTTPRestStreamingResponse finalStreamingResponse = streamingResponse;
-                executor.execute(() -> {
-                    // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
-                    AtomicLong eventIdCounter = new AtomicLong(0);
-                    Multi<String> sseEvents = Multi.createFrom().publisher(finalStreamingResponse.getPublisher())
-                            .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
-                    // Write SSE-formatted strings to HTTP response
-                    MultiSseSupport.writeSseStrings(sseEvents, rc, context);
-                });
+                // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
+                // CRITICAL: Subscribe synchronously to avoid race condition where EventConsumer
+                // starts emitting events before MultiSseSupport subscribes. The executor.execute()
+                // wrapper caused 100-600ms delays before subscription, causing events to be lost.
+                AtomicLong eventIdCounter = new AtomicLong(0);
+                Multi<String> sseEvents = Multi.createFrom().publisher(streamingResponse.getPublisher())
+                        .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
+                // Write SSE-formatted strings to HTTP response
+                MultiSseSupport.writeSseStrings(sseEvents, rc, context);
             }
         }
     }
@@ -431,15 +431,15 @@ public void subscribeToTask(RoutingContext rc) {
             if (error != null) {
                 sendResponse(rc, error);
             } else if (streamingResponse != null) {
-                final HTTPRestStreamingResponse finalStreamingResponse = streamingResponse;
-                executor.execute(() -> {
-                    // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
-                    AtomicLong eventIdCounter = new AtomicLong(0);
-                    Multi<String> sseEvents = Multi.createFrom().publisher(finalStreamingResponse.getPublisher())
-                            .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
-                    // Write SSE-formatted strings to HTTP response
-                    MultiSseSupport.writeSseStrings(sseEvents, rc, context);
-                });
+                // Convert Flow.Publisher<String> (JSON) to Multi<String> (SSE-formatted)
+                // CRITICAL: Subscribe synchronously to avoid race condition where EventConsumer
+                // starts emitting events before MultiSseSupport subscribes. The executor.execute()
+                // wrapper caused 100-600ms delays before subscription, causing events to be lost.
+                AtomicLong eventIdCounter = new AtomicLong(0);
+                Multi<String> sseEvents = Multi.createFrom().publisher(streamingResponse.getPublisher())
+                        .map(json -> SseFormatter.formatJsonAsSSE(json, eventIdCounter.getAndIncrement()));
+                // Write SSE-formatted strings to HTTP response
+                MultiSseSupport.writeSseStrings(sseEvents, rc, context);
             }
         }
     }