Merge pull request #108 from kabir/http-client

Switch out OkHttpClient for HttpClient from the JDK

Author: fjuma

examples/src/main/java/io/a2a/examples/helloworld/server/A2AHttpClientProducer.java

@@ -0,0 +1,17 @@
+package io.a2a.examples.helloworld.server;
+
+import io.a2a.http.A2AHttpClient;
+import io.a2a.http.JdkA2AHttpClient;
+
+import jakarta.enterprise.context.ApplicationScoped;
+import jakarta.enterprise.inject.Produces;
+
+@ApplicationScoped
+public class A2AHttpClientProducer {
+
+    @Produces
+    public A2AHttpClient httpClient() {
+        return new JdkA2AHttpClient();
+    }
+
+}

examples/src/main/java/io/a2a/examples/helloworld/server/TmpA2AHttpClientProducer.java

@@ -25,7 +25,6 @@
         <mockserver.version>5.15.0</mockserver.version>
         <mutiny-zero.version>1.1.1</mutiny-zero.version>
         <quarkus.platform.version>3.22.3</quarkus.platform.version>
-        <okhttp.version>4.12.0</okhttp.version>
         <rest-assured.version>5.5.1</rest-assured.version>
         <slf4j.version>2.0.17</slf4j.version>
 
@@ -63,16 +62,6 @@
             <artifactId>jackson-datatype-jsr310</artifactId>
             <version>${jackson.version}</version>
         </dependency>
-        <dependency>
-            <groupId>com.squareup.okhttp3</groupId>
-            <artifactId>okhttp</artifactId>
-            <version>${okhttp.version}</version>
-        </dependency>
-        <dependency>
-            <groupId>com.squareup.okhttp3</groupId>
-            <artifactId>okhttp-sse</artifactId>
-            <version>${okhttp.version}</version>
-        </dependency>
         <dependency>
             <groupId>io.smallrye.reactive</groupId>
             <artifactId>mutiny-zero</artifactId>

pom.xml

@@ -13,11 +13,16 @@
 
 import java.io.IOException;
 import java.util.Map;
+import java.util.concurrent.CompletableFuture;
+import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.core.type.TypeReference;
 import io.a2a.client.sse.SSEEventListener;
+import io.a2a.http.A2AHttpClient;
+import io.a2a.http.A2AHttpResponse;
+import io.a2a.http.JdkA2AHttpClient;
 import io.a2a.spec.A2A;
 import io.a2a.spec.A2AServerException;
 import io.a2a.spec.AgentCard;
@@ -40,12 +45,6 @@
 import io.a2a.spec.TaskIdParams;
 import io.a2a.spec.TaskPushNotificationConfig;
 import io.a2a.spec.TaskQueryParams;
-import okhttp3.MediaType;
-import okhttp3.OkHttpClient;
-import okhttp3.Request;
-import okhttp3.RequestBody;
-import okhttp3.Response;
-import okhttp3.sse.EventSources;
 
 /**
  * An A2A client.
@@ -57,8 +56,7 @@ public class A2AClient {
     private static final TypeReference<CancelTaskResponse> CANCEL_TASK_RESPONSE_REFERENCE = new TypeReference<>() {};
     private static final TypeReference<GetTaskPushNotificationConfigResponse> GET_TASK_PUSH_NOTIFICATION_CONFIG_RESPONSE_REFERENCE = new TypeReference<>() {};
     private static final TypeReference<SetTaskPushNotificationConfigResponse> SET_TASK_PUSH_NOTIFICATION_CONFIG_RESPONSE_REFERENCE = new TypeReference<>() {};
-    private static final MediaType JSON_MEDIA_TYPE = MediaType.parse("application/json; charset=utf-8");
-    private final OkHttpClient httpClient;
+    private final A2AHttpClient httpClient;
     private final String agentUrl;
     private AgentCard agentCard;
 
@@ -72,7 +70,7 @@ public A2AClient(AgentCard agentCard) {
         checkNotNullParam("agentCard", agentCard);
         this.agentCard = agentCard;
         this.agentUrl = agentCard.url();
-        this.httpClient = new OkHttpClient();
+        this.httpClient = new JdkA2AHttpClient();
     }
 
     /**
@@ -83,7 +81,7 @@ public A2AClient(AgentCard agentCard) {
     public A2AClient(String agentUrl) {
         checkNotNullParam("agentUrl", agentUrl);
         this.agentUrl = agentUrl;
-        this.httpClient = new OkHttpClient();
+        this.httpClient = new JdkA2AHttpClient();
     }
 
     /**
@@ -349,20 +347,6 @@ public SetTaskPushNotificationConfigResponse setTaskPushNotificationConfig(Strin
         }
     }
 
-    /**
-     * Send a streaming message to the remote agent.
-     *
-     * @param messageSendParams the parameters for the message to be sent
-     * @param eventHandler a consumer that will be invoked for each event received from the remote agent
-     * @param errorHandler a consumer that will be invoked if the remote agent returns an error
-     * @param failureHandler a consumer that will be invoked if a failure occurs when processing events
-     * @throws A2AServerException if sending the streaming message fails for any reason
-     */
-    public void sendStreamingMessage(MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler,
-                                     Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException {
-        sendStreamingMessage(null, messageSendParams, eventHandler, errorHandler, failureHandler);
-    }
-
     /**
      * Send a streaming message to the remote agent.
      *
@@ -374,7 +358,7 @@ public void sendStreamingMessage(MessageSendParams messageSendParams, Consumer<S
      * @throws A2AServerException if sending the streaming message fails for any reason
      */
     public void sendStreamingMessage(String requestId, MessageSendParams messageSendParams, Consumer<StreamingEventKind> eventHandler,
-                                     Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException {
+                                       Consumer<JSONRPCError> errorHandler, Runnable failureHandler) throws A2AServerException {
         checkNotNullParam("messageSendParams", messageSendParams);
         checkNotNullParam("eventHandler", eventHandler);
         checkNotNullParam("errorHandler", errorHandler);
@@ -389,52 +373,41 @@ public void sendStreamingMessage(String requestId, MessageSendParams messageSend
             sendStreamingMessageRequestBuilder.id(requestId);
         }
 
+        AtomicReference<CompletableFuture<Void>> ref = new AtomicReference<>();
+        SSEEventListener sseEventListener = new SSEEventListener(eventHandler, errorHandler, failureHandler);
         SendStreamingMessageRequest sendStreamingMessageRequest = sendStreamingMessageRequestBuilder.build();
-        SSEEventListener sseEventListener = new SSEEventListener.Builder()
-                .eventHandler(eventHandler)
-                .errorHandler(errorHandler)
-                .failureHandler(failureHandler)
-                .build();
         try {
-            EventSources.createFactory(httpClient)
-                    .newEventSource(createPostRequest(sendStreamingMessageRequest,
-                            true), sseEventListener);
+            A2AHttpClient.PostBuilder builder = createPostBuilder(sendStreamingMessageRequest);
+            ref.set(builder.postAsyncSSE(
+                    msg -> sseEventListener.onMessage(msg, ref.get()),
+                    throwable -> sseEventListener.onError(throwable, ref.get()),
+                    () -> {
+                        // We don't need to do anything special on completion
+                    }));
+
         } catch (IOException e) {
             throw new A2AServerException("Failed to send streaming message request: " + e);
+        } catch (InterruptedException e) {
+            throw new A2AServerException("Send streaming message request timed out: " + e);
         }
     }
 
-    private String sendPostRequest(Object value) throws IOException, InterruptedException{
-        return sendPostRequest(value, false);
-    }
-
-
-    private String sendPostRequest(Object value, boolean addEventStreamHeader) throws IOException, InterruptedException{
-        Request okRequest = createPostRequest(value, addEventStreamHeader);
-        try (Response response = httpClient.newCall(okRequest).execute()) {
-            if (! response.isSuccessful()) {
-                throw new IOException("Request failed " + response.code());
-            }
-            return response.body().string();
+    private String sendPostRequest(Object value) throws IOException, InterruptedException {
+        A2AHttpClient.PostBuilder builder = createPostBuilder(value);
+        A2AHttpResponse response = builder.post();
+        if (!response.success()) {
+            throw new IOException("Request failed " + response.status());
         }
-
+        return response.body();
     }
 
-    private Request createPostRequest(Object value) throws IOException {
-        return createPostRequest(value, false);
-    }
-
-    private Request createPostRequest(Object value, boolean addEventStreamHeader) throws IOException {
-        Request.Builder builder = new Request.Builder()
+    private A2AHttpClient.PostBuilder createPostBuilder(Object value) throws JsonProcessingException {
+        return httpClient.createPost()
                 .url(agentUrl)
                 .addHeader("Content-Type", "application/json")
-                .post(RequestBody.create(OBJECT_MAPPER.writeValueAsString(value), JSON_MEDIA_TYPE));
-        if (addEventStreamHeader) {
-            builder.addHeader("Accept", "text/event-stream");
-        }
-        return builder.build();
-    }
+                .body(OBJECT_MAPPER.writeValueAsString(value));
 
+    }
 
     private <T extends JSONRPCResponse> T unmarshalResponse(String response, TypeReference<T> typeReference)
             throws A2AServerException, JsonProcessingException {

src/main/java/io/a2a/client/A2AClient.java

@@ -1,85 +1,44 @@
 package io.a2a.client.sse;
 
-import static io.a2a.util.Assert.checkNotNullParam;
 import static io.a2a.util.Utils.OBJECT_MAPPER;
 
+import java.util.concurrent.Future;
 import java.util.function.Consumer;
 
 import com.fasterxml.jackson.core.JsonProcessingException;
 import com.fasterxml.jackson.databind.JsonNode;
 import io.a2a.spec.JSONRPCError;
 import io.a2a.spec.StreamingEventKind;
 import io.a2a.spec.TaskStatusUpdateEvent;
-import okhttp3.Response;
-import okhttp3.sse.EventSource;
-import okhttp3.sse.EventSourceListener;
 import org.slf4j.Logger;
 import org.slf4j.LoggerFactory;
 
-/**
- * A listener for Server-Sent Events (SSE).
- */
-public class SSEEventListener extends EventSourceListener {
-
+public class SSEEventListener {
     private static final Logger log = LoggerFactory.getLogger(SSEEventListener.class);
-    private final boolean logEvents;
     private final Consumer<StreamingEventKind> eventHandler;
     private final Consumer<JSONRPCError> errorHandler;
     private final Runnable failureHandler;
 
-    public SSEEventListener(Consumer<StreamingEventKind> eventHandler, Consumer<JSONRPCError> errorHandler,
-                            Runnable failureHandler) {
-        this(false, eventHandler, errorHandler, failureHandler);
-    }
-
-    public SSEEventListener(boolean logEvents, Consumer<StreamingEventKind> eventHandler,
-                            Consumer<JSONRPCError> errorHandler, Runnable failureHandler) {
-        checkNotNullParam("eventHandler", eventHandler);
-        checkNotNullParam("errorHandler", errorHandler);
-        checkNotNullParam("failureHandler", failureHandler);
-        this.logEvents = logEvents;
+    public SSEEventListener(Consumer<StreamingEventKind> eventHandler, Consumer<JSONRPCError> errorHandler, Runnable failureHandler) {
         this.eventHandler = eventHandler;
         this.errorHandler = errorHandler;
         this.failureHandler = failureHandler;
     }
 
-    @Override
-    public void onOpen(EventSource eventSource, Response response) {
-        if (logEvents) {
-            log.debug("onOpen()");
-        }
-    }
-
-    @Override
-    public void onClosed(EventSource eventSource) {
-        if (logEvents) {
-            log.debug("onClosed()");
-        }
-    }
-
-    @Override
-    public void onEvent(EventSource eventSource, String id, String type, String data) {
-        if (type.equals("message")) {
-            if (logEvents) {
-                log.debug("onEvent() {}", data);
-            }
-            try {
-                handleMessage(OBJECT_MAPPER.readTree(data), eventSource);
-            } catch (JsonProcessingException e) {
-                log.warn("Failed to parse JSON message: {}", data, e);
-            }
+    public void onMessage(String message, Future<Void> completableFuture) {
+        try {
+            handleMessage(OBJECT_MAPPER.readTree(message),completableFuture);
+        } catch (JsonProcessingException e) {
+            log.warn("Failed to parse JSON message: {}", message, e);
         }
     }
 
-    @Override
-    public void onFailure(EventSource eventSource, Throwable t, Response response) {
-        if (logEvents) {
-            log.debug("onFailure()", t);
-        }
+    public void onError(Throwable throwable, Future<Void> future) {
         failureHandler.run();
+        future.cancel(true); // close SSE channel
     }
 
-    private void handleMessage(JsonNode jsonNode, EventSource eventSource) {
+    private void handleMessage(JsonNode jsonNode, Future<Void> future) {
         try {
             if (jsonNode.has("error")) {
                 JSONRPCError error = OBJECT_MAPPER.treeToValue(jsonNode.get("error"), JSONRPCError.class);
@@ -90,7 +49,7 @@ private void handleMessage(JsonNode jsonNode, EventSource eventSource) {
                 StreamingEventKind event = OBJECT_MAPPER.treeToValue(result, StreamingEventKind.class);
                 eventHandler.accept(event);
                 if (event instanceof TaskStatusUpdateEvent && ((TaskStatusUpdateEvent) event).isFinal()) {
-                    eventSource.cancel(); // close SSE channel
+                    future.cancel(true); // close SSE channel
                 }
             } else {
                 throw new IllegalArgumentException("Unknown message type");
@@ -100,34 +59,4 @@ private void handleMessage(JsonNode jsonNode, EventSource eventSource) {
         }
     }
 
-    public static class Builder {
-        private boolean logEvents;
-        private Consumer<StreamingEventKind> eventHandler;
-        private Consumer<JSONRPCError> errorHandler;
-        private Runnable failureHandler;
-
-        public Builder logEvents(boolean logEvents) {
-            this.logEvents = logEvents;
-            return this;
-        }
-
-        public Builder eventHandler(Consumer<StreamingEventKind> eventHandler) {
-            this.eventHandler = eventHandler;
-            return this;
-        }
-
-        public Builder errorHandler(Consumer<JSONRPCError> errorHandler) {
-            this.errorHandler = errorHandler;
-            return this;
-        }
-
-        public Builder failureHandler(Runnable failureHandler) {
-            this.failureHandler = failureHandler;
-            return this;
-        }
-
-        public SSEEventListener build() {
-            return new SSEEventListener(logEvents, eventHandler, errorHandler, failureHandler);
-        }
-    }
 }