Adding streaming send message and setPushNotificationConfig

Signed-off-by: Emmanuel Hugonnet <ehugonne@redhat.com>

Author: ehsavoie

client/pom.xml

@@ -46,6 +46,11 @@
             <artifactId>protobuf-java-util</artifactId>
             <type>jar</type>
         </dependency>
+        <dependency>
+            <groupId>org.slf4j</groupId>
+            <artifactId>slf4j-jdk14</artifactId>
+            <scope>test</scope>
+        </dependency>
     </dependencies>
 
 </project>

client/src/main/java/io/a2a/client/sse/JSONRestSSEEventListener.java

@@ -0,0 +1,72 @@
+package io.a2a.client.sse;
+
+import static io.a2a.grpc.StreamResponse.PayloadCase.ARTIFACT_UPDATE;
+import static io.a2a.grpc.StreamResponse.PayloadCase.MSG;
+import static io.a2a.grpc.StreamResponse.PayloadCase.STATUS_UPDATE;
+import static io.a2a.grpc.StreamResponse.PayloadCase.TASK;
+
+import java.util.concurrent.Future;
+import java.util.function.Consumer;
+import java.util.logging.Logger;
+
+import com.google.protobuf.InvalidProtocolBufferException;
+import com.google.protobuf.util.JsonFormat;
+import io.a2a.grpc.StreamResponse;
+import io.a2a.grpc.utils.ProtoUtils;
+import io.a2a.spec.StreamingEventKind;
+import java.util.logging.Level;
+
+public class JSONRestSSEEventListener {
+
+    private static final Logger log = Logger.getLogger(JSONRestSSEEventListener.class.getName());
+    private final Consumer<StreamingEventKind> eventHandler;
+    private final Consumer<Throwable> errorHandler;
+
+    public JSONRestSSEEventListener(Consumer<StreamingEventKind> eventHandler,
+            Consumer<Throwable> errorHandler) {
+        this.eventHandler = eventHandler;
+        this.errorHandler = errorHandler;
+    }
+
+    public void onMessage(String message, Future<Void> completableFuture) {
+        try {
+            io.a2a.grpc.StreamResponse.Builder builder = io.a2a.grpc.StreamResponse.newBuilder();
+            JsonFormat.parser().merge(message, builder);
+            handleMessage(builder.build(), completableFuture);
+        } catch (InvalidProtocolBufferException e) {
+            Logger.getLogger(JSONRestSSEEventListener.class.getName()).log(Level.SEVERE, null, e);
+            errorHandler.accept(e);
+        }
+    }
+
+    public void onError(Throwable throwable, Future<Void> future) {
+        if (errorHandler != null) {
+            errorHandler.accept(throwable);
+        }
+        future.cancel(true); // close SSE channel
+    }
+
+    private void handleMessage(StreamResponse response, Future<Void> future) {
+        StreamingEventKind event;
+        switch (response.getPayloadCase()) {
+            case MSG:
+                event = ProtoUtils.FromProto.message(response.getMsg());
+                break;
+            case TASK:
+                event = ProtoUtils.FromProto.task(response.getTask());
+                break;
+            case STATUS_UPDATE:
+                event = ProtoUtils.FromProto.taskStatusUpdateEvent(response.getStatusUpdate());
+                break;
+            case ARTIFACT_UPDATE:
+                event = ProtoUtils.FromProto.taskArtifactUpdateEvent(response.getArtifactUpdate());
+                break;
+            default:
+                log.warning("Invalid stream response " + response.getPayloadCase());
+                errorHandler.accept(new IllegalStateException("Invalid stream response from server: " + response.getPayloadCase()));
+                return;
+        }
+        eventHandler.accept(event);
+    }
+
+}

client/src/main/java/io/a2a/client/transport/JSONRestTransport.java

@@ -0,0 +1,22 @@
+package io.a2a.client.transport;
+
+import java.util.List;
+
+import io.a2a.client.ClientCallInterceptor;
+import io.a2a.client.ClientConfig;
+import io.a2a.spec.AgentCard;
+import io.a2a.spec.TransportProtocol;
+
+public class JSONRestTransportProvider implements ClientTransportProvider {
+
+    @Override
+    public ClientTransport create(ClientConfig clientConfig, AgentCard agentCard,
+                                  String agentUrl, List<ClientCallInterceptor> interceptors) {
+        return new JSONRestTransport(clientConfig.getHttpClient(), agentCard, agentUrl, interceptors);
+    }
+
+    @Override
+    public String getTransportProtocol() {
+        return TransportProtocol.HTTP_JSON.asString();
+    }
+}

client/src/main/java/io/a2a/client/transport/JSONRestTransportProvider.java

@@ -10,6 +10,8 @@ public interface A2AHttpClient {
 
     PostBuilder createPost();
 
+    DeleteBuilder createDelete();
+
     interface Builder<T extends Builder<T>> {
         T url(String s);
         T addHeader(String name, String value);
@@ -31,4 +33,8 @@ CompletableFuture<Void> postAsyncSSE(
                 Consumer<Throwable> errorConsumer,
                 Runnable completeRunnable) throws IOException, InterruptedException;
     }
+
+    interface DeleteBuilder extends Builder<DeleteBuilder> {
+        A2AHttpResponse delete() throws IOException, InterruptedException;
+    }
 }

client/src/main/java/io/a2a/http/A2AHttpClient.java

@@ -35,6 +35,11 @@ public PostBuilder createPost() {
         return new JdkPostBuilder();
     }
 
+    @Override
+    public DeleteBuilder createDelete() {
+        return new JdkDeleteBuilder();
+    }
+
     private abstract class JdkBuilder<T extends Builder<T>> implements Builder<T> {
         private String url;
         private Map<String, String> headers = new HashMap<>();
@@ -147,6 +152,18 @@ public CompletableFuture<Void> getAsyncSSE(
         }
     }
 
+    private class JdkDeleteBuilder extends JdkBuilder<DeleteBuilder> implements A2AHttpClient.DeleteBuilder {
+
+        @Override
+        public A2AHttpResponse delete() throws IOException, InterruptedException {
+            HttpRequest request = super.createRequestBuilder().DELETE().build();
+            HttpResponse<String> response =
+                    httpClient.send(request, BodyHandlers.ofString(StandardCharsets.UTF_8));
+            return new JdkHttpResponse(response);
+        }
+
+    }
+
     private class JdkPostBuilder extends JdkBuilder<PostBuilder> implements A2AHttpClient.PostBuilder {
         String body = "";
 

client/src/main/java/io/a2a/http/JdkA2AHttpClient.java

@@ -1,2 +1,3 @@
 io.a2a.client.transport.JSONRPCTransportProvider
-io.a2a.client.transport.GrpcTransportProvider
+io.a2a.client.transport.GrpcTransportProvider
+io.a2a.client.transport.JSONRestTransportProvider

client/src/main/resources/META-INF/services/io.a2a.client.transport.ClientTransportProvider

@@ -128,6 +128,11 @@ public PostBuilder createPost() {
             return null;
         }
 
+        @Override
+        public DeleteBuilder createDelete() {
+            return null;
+        }
+
         class TestGetBuilder implements A2AHttpClient.GetBuilder {
 
             @Override