Merge pull request #121 from kabir/one-executor

Share the Executor used for async operations

Author: maeste

core/src/main/java/io/a2a/server/requesthandlers/DefaultRequestHandler.java

@@ -1,8 +1,8 @@
 package io.a2a.server.requesthandlers;
 
-import static io.a2a.util.AsyncUtils.convertingProcessor;
-import static io.a2a.util.AsyncUtils.createTubeConfig;
-import static io.a2a.util.AsyncUtils.processor;
+import static io.a2a.util.async.AsyncUtils.convertingProcessor;
+import static io.a2a.util.async.AsyncUtils.createTubeConfig;
+import static io.a2a.util.async.AsyncUtils.processor;
 
 import java.util.ArrayList;
 import java.util.Collections;
@@ -12,7 +12,6 @@
 import java.util.Objects;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Supplier;
@@ -47,6 +46,7 @@
 import io.a2a.spec.TaskQueryParams;
 import io.a2a.spec.UnsupportedOperationError;
 import io.a2a.util.TempLoggerWrapper;
+import io.a2a.util.async.Internal;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
@@ -63,15 +63,16 @@ public class DefaultRequestHandler implements RequestHandler {
 
     private final Map<String, CompletableFuture<Void>> runningAgents = Collections.synchronizedMap(new HashMap<>());
 
-    private final Executor executor = Executors.newCachedThreadPool();
+    private final Executor executor;
 
     @Inject
     public DefaultRequestHandler(AgentExecutor agentExecutor, TaskStore taskStore,
-                                 QueueManager queueManager, PushNotifier pushNotifier) {
+                                 QueueManager queueManager, PushNotifier pushNotifier, @Internal Executor executor) {
         this.agentExecutor = agentExecutor;
         this.taskStore = taskStore;
         this.queueManager = queueManager;
         this.pushNotifier = pushNotifier;
+        this.executor = executor;
         // TODO In Python this is also a constructor parameter defaulting to this SimpleRequestContextBuilder
         //  implementation if the parameter is null. Skip that for now, since otherwise I get CDI errors, and
         //  I am unsure about the correct scope.

core/src/main/java/io/a2a/server/requesthandlers/JSONRPCHandler.java

@@ -1,6 +1,6 @@
 package io.a2a.server.requesthandlers;
 
-import static io.a2a.util.AsyncUtils.createTubeConfig;
+import static io.a2a.util.async.AsyncUtils.createTubeConfig;
 
 import java.util.concurrent.Flow;
 

core/src/main/java/io/a2a/server/tasks/ResultAggregator.java

@@ -1,8 +1,8 @@
 package io.a2a.server.tasks;
 
-import static io.a2a.util.AsyncUtils.consumer;
-import static io.a2a.util.AsyncUtils.createTubeConfig;
-import static io.a2a.util.AsyncUtils.processor;
+import static io.a2a.util.async.AsyncUtils.consumer;
+import static io.a2a.util.async.AsyncUtils.createTubeConfig;
+import static io.a2a.util.async.AsyncUtils.processor;
 
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;

core/src/main/java/io/a2a/util/async/AsyncExecutorProducer.java

@@ -0,0 +1,33 @@
+package io.a2a.util.async;
+
+import java.util.concurrent.Executor;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+
+import jakarta.annotation.PostConstruct;
+import jakarta.annotation.PreDestroy;
+import jakarta.enterprise.inject.Produces;
+import jakarta.inject.Singleton;
+
+@Singleton
+public class AsyncExecutorProducer {
+
+    private ExecutorService executor;
+
+    @PostConstruct
+    public void init() {
+        executor = Executors.newCachedThreadPool();
+    }
+
+    @PreDestroy
+    public void close() {
+        executor.shutdown();
+    }
+
+    @Produces
+    @Internal
+    public Executor produce() {
+        return executor;
+    }
+
+}

…rc/main/java/io/a2a/util/AsyncUtils.java …n/java/io/a2a/util/async/AsyncUtils.javacore/src/main/java/io/a2a/util/AsyncUtils.java renamed to core/src/main/java/io/a2a/util/async/AsyncUtils.java core/src/main/java/io/a2a/util/AsyncUtils.java renamed to core/src/main/java/io/a2a/util/async/AsyncUtils.java

@@ -1,4 +1,4 @@
-package io.a2a.util;
+package io.a2a.util.async;
 
 import java.util.concurrent.Flow;
 import java.util.concurrent.atomic.AtomicBoolean;
@@ -7,6 +7,7 @@
 import java.util.function.Consumer;
 import java.util.function.Function;
 
+import io.a2a.util.Assert;
 import mutiny.zero.BackpressureStrategy;
 import mutiny.zero.Tube;
 import mutiny.zero.TubeConfiguration;

core/src/main/java/io/a2a/util/async/Internal.java

@@ -0,0 +1,12 @@
+package io.a2a.util.async;
+
+import java.lang.annotation.Retention;
+import java.lang.annotation.RetentionPolicy;
+import java.lang.annotation.Target;
+
+import jakarta.inject.Qualifier;
+
+@Qualifier
+@Retention(RetentionPolicy.RUNTIME)
+public @interface Internal {
+}

core/src/test/java/io/a2a/server/requesthandlers/JSONRPCHandlerTest.java

@@ -14,6 +14,7 @@
 import java.util.List;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
+import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
 import java.util.concurrent.TimeUnit;
@@ -104,6 +105,8 @@ public class JSONRPCHandlerTest {
     private InMemoryQueueManager queueManager;
     private TestHttpClient httpClient;
 
+    private final Executor internalExecutor = Executors.newCachedThreadPool();
+
 
     @BeforeEach
     public void init() {
@@ -128,7 +131,7 @@ public void cancel(RequestContext context, EventQueue eventQueue) throws JSONRPC
         httpClient = new TestHttpClient();
         PushNotifier pushNotifier = new InMemoryPushNotifier(httpClient);
 
-        requestHandler = new DefaultRequestHandler(executor, taskStore, queueManager, pushNotifier);
+        requestHandler = new DefaultRequestHandler(executor, taskStore, queueManager, pushNotifier, internalExecutor);
     }
 
     @AfterEach
@@ -1050,7 +1053,8 @@ public void testPushNotificationsNotSupportedError() {
     @Test
     public void testOnGetPushNotificationNoPushNotifier() {
         // Create request handler without a push notifier
-        DefaultRequestHandler requestHandler = new DefaultRequestHandler(executor, taskStore, queueManager, null);
+        DefaultRequestHandler requestHandler =
+                new DefaultRequestHandler(executor, taskStore, queueManager, null, internalExecutor);
         AgentCard card = createAgentCard(false, true, false);
         JSONRPCHandler handler = new JSONRPCHandler(card, requestHandler);
 
@@ -1068,7 +1072,8 @@ public void testOnGetPushNotificationNoPushNotifier() {
     @Test
     public void testOnSetPushNotificationNoPushNotifier() {
         // Create request handler without a push notifier
-        DefaultRequestHandler requestHandler = new DefaultRequestHandler(executor, taskStore, queueManager, null);
+        DefaultRequestHandler requestHandler =
+                new DefaultRequestHandler(executor, taskStore, queueManager, null, internalExecutor);
         AgentCard card = createAgentCard(false, true, false);
         JSONRPCHandler handler = new JSONRPCHandler(card, requestHandler);
 
@@ -1157,7 +1162,8 @@ public void testDefaultRequestHandlerWithCustomComponents() {
 
     @Test
     public void testOnMessageSendErrorHandling() {
-        DefaultRequestHandler requestHandler = new DefaultRequestHandler(executor, taskStore, queueManager, null);
+        DefaultRequestHandler requestHandler =
+                new DefaultRequestHandler(executor, taskStore, queueManager, null, internalExecutor);
         AgentCard card = createAgentCard(false, true, false);
         JSONRPCHandler handler = new JSONRPCHandler(card, requestHandler);
 

core/src/test/java/io/a2a/util/AsyncUtilsTest.java

@@ -1,9 +1,9 @@
 package io.a2a.util;
 
-import static io.a2a.util.AsyncUtils.consumer;
-import static io.a2a.util.AsyncUtils.convertingProcessor;
-import static io.a2a.util.AsyncUtils.createTubeConfig;
-import static io.a2a.util.AsyncUtils.processor;
+import static io.a2a.util.async.AsyncUtils.consumer;
+import static io.a2a.util.async.AsyncUtils.convertingProcessor;
+import static io.a2a.util.async.AsyncUtils.createTubeConfig;
+import static io.a2a.util.async.AsyncUtils.processor;
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
 import static org.junit.jupiter.api.Assertions.assertNotNull;
@@ -13,8 +13,6 @@
 import java.util.Collections;
 import java.util.List;
 import java.util.concurrent.CountDownLatch;
-import java.util.concurrent.Executor;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;

sdk-jakarta/src/main/java/io/a2a/server/apps/jakarta/A2AServerResource.java

@@ -47,6 +47,7 @@
 import io.a2a.spec.StreamingJSONRPCRequest;
 import io.a2a.spec.TaskResubscriptionRequest;
 import io.a2a.spec.UnsupportedOperationError;
+import io.a2a.util.async.Internal;
 
 @Path("/")
 public class A2AServerResource {
@@ -61,8 +62,9 @@ public class A2AServerResource {
     // Hook so testing can wait until the async Subscription is subscribed.
     private static volatile Runnable streamingIsSubscribedRunnable;
 
-
-    private final Executor executor = Executors.newCachedThreadPool();
+    @Inject
+    @Internal
+    Executor executor;
 
     /**
      * Handles incoming POST requests to the main A2A endpoint. Dispatches the

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

@@ -45,6 +45,7 @@
 import io.a2a.spec.TaskResubscriptionRequest;
 import io.a2a.spec.UnsupportedOperationError;
 import io.a2a.util.Utils;
+import io.a2a.util.async.Internal;
 import io.quarkus.vertx.web.Body;
 import io.quarkus.vertx.web.ReactiveRoutes;
 import io.quarkus.vertx.web.Route;
@@ -71,7 +72,9 @@ public class A2AServerRoutes {
     // Hook so testing can wait until the MultiSseSupport is subscribed.
     private static volatile Runnable streamingMultiSseSupportSubscribedRunnable;
 
-    private final Executor executor = Executors.newCachedThreadPool();
+    @Inject
+    @Internal
+    Executor executor;
 
     @Route(path = "/", methods = {Route.HttpMethod.POST}, consumes = {APPLICATION_JSON}, type = Route.HandlerType.BLOCKING)
     public void invokeJSONRPCHandler(@Body String body, RoutingContext rc) {