Move QueueManager interactions to REST

Author: kabir

reference-impl/src/test/java/io/a2a/server/apps/quarkus/A2ATestRoutes.java

@@ -9,12 +9,17 @@
 
 import io.a2a.server.apps.common.TestUtilsBean;
 import io.a2a.spec.Task;
+import io.a2a.spec.TaskArtifactUpdateEvent;
+import io.a2a.spec.TaskStatusUpdateEvent;
 import io.a2a.util.Utils;
 import io.quarkus.vertx.web.Body;
 import io.quarkus.vertx.web.Param;
 import io.quarkus.vertx.web.Route;
 import io.vertx.ext.web.RoutingContext;
 
+/**
+ * Exposes the {@link TestUtilsBean} via REST using Quarkus Reactive Routes
+ */
 @Singleton
 public class A2ATestRoutes {
     @Inject
@@ -73,6 +78,46 @@ public void deleteTask(@Param String taskId, RoutingContext rc) {
         }
     }
 
+    @Route(path = "test/queue/ensure/:taskId", methods = {Route.HttpMethod.POST})
+    public void ensureTaskQueue(@Param String taskId, RoutingContext rc) {
+        try {
+            testUtilsBean.ensureQueue(taskId);
+            rc.response()
+                    .setStatusCode(200)
+                    .end();
+        } catch (Throwable t) {
+            errorResponse(t, rc);
+        }
+    }
+
+    @Route(path = "test/queue/enqueueTaskStatusUpdateEvent/:taskId", methods = {Route.HttpMethod.POST})
+    public void enqueueTaskStatusUpdateEvent(@Param String taskId, @Body String body, RoutingContext rc) {
+
+        try {
+            TaskStatusUpdateEvent event = Utils.OBJECT_MAPPER.readValue(body, TaskStatusUpdateEvent.class);
+            testUtilsBean.enqueueEvent(taskId, event);
+            rc.response()
+                    .setStatusCode(200)
+                    .end();
+        } catch (Throwable t) {
+            errorResponse(t, rc);
+        }
+    }
+
+    @Route(path = "test/queue/enqueueTaskArtifactUpdateEvent/:taskId", methods = {Route.HttpMethod.POST})
+    public void enqueueTaskArtifactUpdateEvent(@Param String taskId, @Body String body, RoutingContext rc) {
+
+        try {
+            TaskArtifactUpdateEvent event = Utils.OBJECT_MAPPER.readValue(body, TaskArtifactUpdateEvent.class);
+            testUtilsBean.enqueueEvent(taskId, event);
+            rc.response()
+                    .setStatusCode(200)
+                    .end();
+        } catch (Throwable t) {
+            errorResponse(t, rc);
+        }
+    }
+
     private void errorResponse(Throwable t, RoutingContext rc) {
         rc.response()
                 .setStatusCode(200)

reference-impl/src/test/java/io/a2a/server/apps/quarkus/QuarkusA2AServerTest.java

@@ -10,18 +10,10 @@
 @QuarkusTest
 public class QuarkusA2AServerTest extends AbstractA2AServerTest {
 
-    @Inject
-    InMemoryQueueManager queueManager;
-
     public QuarkusA2AServerTest() {
         super(8081);
     }
 
-    @Override
-    protected InMemoryQueueManager getQueueManager() {
-        return queueManager;
-    }
-
     @Override
     protected void setStreamingSubscribedRunnable(Runnable runnable) {
         A2AServerRoutes.setStreamingMultiSseSupportSubscribedRunnable(runnable);

tests/server-common/src/test/java/io/a2a/server/apps/common/AbstractA2AServerTest.java

@@ -630,7 +630,7 @@ public void testResubscribeExistingTaskSuccess() throws Exception {
             // attempting to send a streaming message instead of explicitly calling queueManager#createOrTap
             // does not work because after the message is sent, the queue becomes null but task resubscription
             // requires the queue to still be active
-            getQueueManager().createOrTap(MINIMAL_TASK.getId());
+            ensureQueueForTask(MINIMAL_TASK.getId());
 
             CountDownLatch taskResubscriptionRequestSent = new CountDownLatch(1);
             CountDownLatch taskResubscriptionResponseReceived = new CountDownLatch(2);
@@ -701,7 +701,7 @@ public void testResubscribeExistingTaskSuccess() throws Exception {
                                 .build());
 
                 for (Event event : events) {
-                    getQueueManager().get(MINIMAL_TASK.getId()).enqueueEvent(event);
+                    enqueueEventOnServer(event);
                 }
             } catch (InterruptedException e) {
                 Thread.currentThread().interrupt();
@@ -941,7 +941,43 @@ protected void deleteTaskInTaskStore(String taskId) throws Exception {
         }
     }
 
-    protected abstract InMemoryQueueManager getQueueManager();
+    protected void ensureQueueForTask(String taskId) throws Exception {
+        HttpClient client = HttpClient.newBuilder()
+                .version(HttpClient.Version.HTTP_2)
+                .build();
+        HttpRequest request = HttpRequest.newBuilder()
+                .uri(URI.create("http://localhost:" + serverPort + "/test/queue/ensure/" + taskId))
+                .POST(HttpRequest.BodyPublishers.noBody())
+                .build();
+        HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
+        if (response.statusCode() != 200) {
+            throw new RuntimeException(response.statusCode() + ": Deleting task failed!" + response.body());
+        }
+    }
+
+    protected void enqueueEventOnServer(Event event) throws Exception {
+        String path = null;
+        if (event instanceof TaskArtifactUpdateEvent e) {
+            path = "test/queue/enqueueTaskArtifactUpdateEvent/" + e.getTaskId();
+        } else if (event instanceof TaskStatusUpdateEvent e) {
+            path = "test/queue/enqueueTaskStatusUpdateEvent/" + e.getTaskId();
+        } else {
+            throw new RuntimeException("Unknown event type " + event.getClass() + ". If you need the ability to" +
+                    " handle more types, please add the REST endpoints.");
+        }
+        HttpClient client = HttpClient.newBuilder()
+                .version(HttpClient.Version.HTTP_2)
+                .build();
+        HttpRequest request = HttpRequest.newBuilder()
+                .uri(URI.create("http://localhost:" + serverPort + "/" + path))
+                .POST(HttpRequest.BodyPublishers.ofString(Utils.OBJECT_MAPPER.writeValueAsString(event)))
+                .build();
+
+        HttpResponse<String> response = client.send(request, HttpResponse.BodyHandlers.ofString(StandardCharsets.UTF_8));
+        if (response.statusCode() != 200) {
+            throw new RuntimeException(response.statusCode() + ": Deleting task failed!" + response.body());
+        }
+    }
 
     protected abstract void setStreamingSubscribedRunnable(Runnable runnable);