Switch InMemoryPushNotifier to use the new client

Author: kabir

src/main/java/io/a2a/http/TempA2AHttpClient.java

@@ -1,23 +1,26 @@
 package io.a2a.server.tasks;
 
+import java.io.IOException;
 import java.util.Collections;
 import java.util.HashMap;
 import java.util.Map;
 
 import jakarta.enterprise.context.ApplicationScoped;
 import jakarta.inject.Inject;
 
-import io.a2a.http.TempA2AHttpClient;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import io.a2a.http.A2AHttpClient;
 import io.a2a.spec.PushNotificationConfig;
 import io.a2a.spec.Task;
+import io.a2a.util.Utils;
 
 @ApplicationScoped
 public class InMemoryPushNotifier implements PushNotifier {
-    private final TempA2AHttpClient httpClient;
+    private final A2AHttpClient httpClient;
     private final Map<String, PushNotificationConfig> pushNotificationInfos = Collections.synchronizedMap(new HashMap<>());
 
     @Inject
-    public InMemoryPushNotifier(TempA2AHttpClient httpClient) {
+    public InMemoryPushNotifier(A2AHttpClient httpClient) {
         this.httpClient = httpClient;
     }
 
@@ -43,8 +46,28 @@ public void sendNotification(Task task) {
             return;
         }
         String url = pushInfo.url();
-        // TODO https://github.com/fjuma/a2a-java-sdk/issues/59 will have the real client
-        httpClient.post(url, task);
+
+        // TODO auth
+
+        String body;
+        try {
+            body = Utils.OBJECT_MAPPER.writeValueAsString(task);
+        } catch (JsonProcessingException e) {
+            e.printStackTrace();
+            throw new RuntimeException("Error writing value as string: " + e.getMessage(), e);
+        } catch (Throwable throwable) {
+            throwable.printStackTrace();
+            throw new RuntimeException("Error writing value as string: " + throwable.getMessage(), throwable);
+        }
+
+        try {
+            httpClient.createPost()
+                    .url(url)
+                    .body(body)
+                    .post();
+        } catch (IOException | InterruptedException e) {
+            throw new RuntimeException("Error pushing data to " + url + ": " + e.getMessage(), e);
+        }
 
     }
 }

src/main/java/io/a2a/server/tasks/InMemoryPushNotifier.java

@@ -8,18 +8,22 @@
 import static org.junit.jupiter.api.Assertions.assertTrue;
 import static org.junit.jupiter.api.Assertions.fail;
 
+import java.io.IOException;
 import java.util.ArrayList;
 import java.util.Collections;
 import java.util.List;
+import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CountDownLatch;
 import java.util.concurrent.Executors;
 import java.util.concurrent.Flow;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicReference;
+import java.util.function.Consumer;
 
 import jakarta.enterprise.context.Dependent;
 
-import io.a2a.http.TempA2AHttpClient;
+import io.a2a.http.A2AHttpClient;
+import io.a2a.http.A2AHttpResponse;
 import io.a2a.server.agentexecution.AgentExecutor;
 import io.a2a.server.agentexecution.RequestContext;
 import io.a2a.server.events.Event;
@@ -65,6 +69,7 @@
 import io.a2a.spec.TaskStatusUpdateEvent;
 import io.a2a.spec.TextPart;
 import io.a2a.spec.UnsupportedOperationError;
+import io.a2a.util.Utils;
 import mutiny.zero.ZeroPublisher;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
@@ -698,7 +703,7 @@ public void onSubscribe(Flow.Subscription subscription) {
 
                 @Override
                 public void onNext(SendStreamingMessageResponse item) {
-                    // System.out.println("-> " + item.getResult());
+                    System.out.println("-> " + item.getResult());
                     results.add(item.getResult());
                     System.out.println(results);
                     subscriptionRef.get().request(1);
@@ -1260,18 +1265,67 @@ private interface AgentExecutorMethod {
     }
 
     @Dependent
-    private static class TestHttpClient implements TempA2AHttpClient {
+    private static class TestHttpClient implements A2AHttpClient {
         final List<Task> tasks = Collections.synchronizedList(new ArrayList<>());
         volatile CountDownLatch latch;
 
         @Override
-        public int post(String url, Task task) {
-            // System.out.println("----> adding " + task);
-            tasks.add(task);
-            if (latch != null) {
-                latch.countDown();
+        public GetBuilder createGet() {
+            return null;
+        }
+
+        @Override
+        public PostBuilder createPost() {
+            return new TestPostBuilder();
+        }
+
+        class TestPostBuilder implements A2AHttpClient.PostBuilder {
+            private volatile String body;
+            @Override
+            public PostBuilder body(String body) {
+                this.body = body;
+                return this;
+            }
+
+            @Override
+            public A2AHttpResponse post() throws IOException, InterruptedException {
+                tasks.add(Utils.OBJECT_MAPPER.readValue(body, Task.TYPE_REFERENCE));
+                try {
+                    return new A2AHttpResponse() {
+                        @Override
+                        public int status() {
+                            return 200;
+                        }
+
+                        @Override
+                        public boolean success() {
+                            return true;
+                        }
+
+                        @Override
+                        public String body() {
+                            return "";
+                        }
+                    };
+                } finally {
+                    latch.countDown();
+                }
+            }
+
+            @Override
+            public CompletableFuture<Void> postAsyncSSE(Consumer<String> messageConsumer, Consumer<Throwable> errorConsumer, Runnable completeRunnable) throws IOException, InterruptedException {
+                return null;
+            }
+
+            @Override
+            public PostBuilder url(String s) {
+                return this;
+            }
+
+            @Override
+            public PostBuilder addHeader(String name, String value) {
+                return this;
             }
-            return 200;
         }
     }
 }

src/test/java/io/a2a/server/requesthandlers/JSONRPCHandlerTest.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();
+    }
+
+}