Add hook to make sure subscrber has started

Tests which need to send events can use this to make sure
there is a subscription before sending events on the queue

This should make testResubscribeExistingTaskSuccess(0 more stable

Author: kabir

quarkus-sdk/src/main/java/io/a2a/server/apps/quarkus/A2AServerRoutes.java

@@ -68,6 +68,9 @@ public class A2AServerRoutes {
     @ExtendedAgentCard
     Instance<AgentCard> extendedAgentCard;
 
+    // Hook so testing can wait until the MultiSseSupport is subscribes.
+    private static volatile Runnable streamingMultiSseSupportSubscribedRunnable;
+
     private final Executor executor = Executors.newCachedThreadPool();
 
     @Route(path = "/", methods = {Route.HttpMethod.POST}, consumes = {APPLICATION_JSON}, type = Route.HandlerType.BLOCKING)
@@ -221,6 +224,10 @@ private static boolean isNonStreamingRequest(String requestBody) {
                 requestBody.contains(A2A.GET_TASK_PUSH_NOTIFICATION_CONFIG_METHOD);
     }
 
+    static void setStreamingMultiSseSupportSubscribedRunnable(Runnable runnable) {
+        streamingMultiSseSupportSubscribedRunnable = runnable;
+    }
+
     // Port of import io.quarkus.vertx.web.runtime.MultiSseSupport, which is considered internal API
     private static class MultiSseSupport {
 
@@ -255,6 +262,12 @@ public static void write(Multi<Buffer> multi, RoutingContext rc) {
                 public void onSubscribe(Flow.Subscription subscription) {
                     this.upstream = subscription;
                     this.upstream.request(1);
+
+                    // Notify tests that we are subscribed
+                    Runnable runnable = streamingMultiSseSupportSubscribedRunnable;
+                    if (runnable != null) {
+                        runnable.run();
+                    }
                 }
 
                 @Override

quarkus-sdk/src/test/java/io/a2a/server/apps/quarkus/A2AServerResourceTest.java

@@ -623,13 +623,15 @@ public void testResubscribeExistingTaskSuccess() throws Exception {
             // resubscribe to the task, requires the task and its queue to still be active
             TaskResubscriptionRequest taskResubscriptionRequest = new TaskResubscriptionRequest("1", new TaskIdParams(MINIMAL_TASK.getId()));
 
+            // Count down the latch when the MultiSseSupport on the server has started subscribing
+            A2AServerRoutes.setStreamingMultiSseSupportSubscribedRunnable(taskResubscriptionRequestSent::countDown);
+
             CompletableFuture<HttpResponse<Stream<String>>> responseFuture = initialiseStreamingRequest(taskResubscriptionRequest, null);
 
-            //CountDownLatch latch = new CountDownLatch(1);
             AtomicReference<Throwable> errorRef = new AtomicReference<>();
-            taskResubscriptionRequestSent.countDown();
 
             responseFuture.thenAccept(response -> {
+
                 if (response.statusCode() != 200) {
                     //errorRef.set(new IllegalStateException("Status code was " + response.statusCode()));
                     throw new IllegalStateException("Status code was " + response.statusCode());
@@ -660,14 +662,11 @@ public void testResubscribeExistingTaskSuccess() throws Exception {
                 if (!isStreamClosedError(t)) {
                     errorRef.set(t);
                 }
-                //latch.countDown();
                 return null;
             });
 
             try {
                 taskResubscriptionRequestSent.await();
-                // sleep to ensure that the events are sent after the client request is made
-                Thread.sleep(1000);
                 List<Event> events = List.of(
                         new TaskArtifactUpdateEvent.Builder()
                                 .taskId(MINIMAL_TASK.getId())
@@ -713,6 +712,7 @@ public void testResubscribeExistingTaskSuccess() throws Exception {
             assertEquals(TaskState.COMPLETED, taskStatusUpdateEvent.getStatus().state());
             assertNotNull(taskStatusUpdateEvent.getStatus().timestamp());
         } finally {
+            A2AServerRoutes.setStreamingMultiSseSupportSubscribedRunnable(null);
             taskStore.delete(MINIMAL_TASK.getId());
             executorService.shutdown();
             if (!executorService.awaitTermination(10, TimeUnit.SECONDS)) {