chore: Update resubscribe and sendMessage without consumers will use … (#528)

…the client's consumers

and its streamingErrorHandler

The code was already doing that down the path but this simplifies the
code as the consumers and streamingErrorhandler can be validated from
the top of the API

Add a bunch of @nullable & @nonnull to clarify the API contract

This fixes #526

Signed-off-by: Jeff Mesnil <[email protected]>

Author: jmesnil

client/base/src/main/java/io/a2a/client/AbstractClient.java

@@ -32,10 +32,10 @@
  */
 public abstract class AbstractClient {
 
-    private final List<BiConsumer<ClientEvent, AgentCard>> consumers;
-    private final @Nullable Consumer<Throwable> streamingErrorHandler;
+    protected final @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers;
+    protected final @Nullable Consumer<Throwable> streamingErrorHandler;
 
-    public AbstractClient(List<BiConsumer<ClientEvent, AgentCard>> consumers) {
+    public AbstractClient(@NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers) {
         this(consumers, null);
     }
 
@@ -57,7 +57,7 @@ public AbstractClient(@NonNull List<BiConsumer<ClientEvent, AgentCard>> consumer
      * @param request the message
      * @throws A2AClientException if sending the message fails for any reason
      */
-    public void sendMessage(Message request) throws A2AClientException {
+    public void sendMessage(@NonNull Message request) throws A2AClientException {
         sendMessage(request, null);
     }
 
@@ -71,10 +71,13 @@ public void sendMessage(Message request) throws A2AClientException {
      * configuration will get used for streaming.
      *
      * @param request the message
-     * @param context optional client call context for the request (may be {@code null})
+     * @param context optional client call context for the request
      * @throws A2AClientException if sending the message fails for any reason
      */
-    public abstract void sendMessage(Message request, @Nullable ClientCallContext context) throws A2AClientException;
+    public void sendMessage(@NonNull Message request,
+                            @Nullable ClientCallContext context) throws A2AClientException {
+        sendMessage(request, consumers, streamingErrorHandler, context);
+    }
 
     /**
      * Send a message to the remote agent. This method will automatically use
@@ -90,9 +93,9 @@ public void sendMessage(Message request) throws A2AClientException {
      * @param streamingErrorHandler an error handler that should be used for the streaming case if an error occurs
      * @throws A2AClientException if sending the message fails for any reason
      */
-    public void sendMessage(Message request,
-                            List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                            Consumer<Throwable> streamingErrorHandler) throws A2AClientException {
+    public void sendMessage(@NonNull Message request,
+                            @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                            @Nullable Consumer<Throwable> streamingErrorHandler) throws A2AClientException {
         sendMessage(request, consumers, streamingErrorHandler, null);
     }
 
@@ -108,12 +111,12 @@ public void sendMessage(Message request,
      * @param request the message
      * @param consumers a list of consumers to pass responses from the remote agent to
      * @param streamingErrorHandler an error handler that should be used for the streaming case if an error occurs
-     * @param context optional client call context for the request (may be {@code null})
+     * @param context optional client call context for the request
      * @throws A2AClientException if sending the message fails for any reason
      */
-    public abstract void sendMessage(Message request,
-                                     List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                                     Consumer<Throwable> streamingErrorHandler,
+    public abstract void sendMessage(@NonNull Message request,
+                                     @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                                     @Nullable Consumer<Throwable> streamingErrorHandler,
                                      @Nullable ClientCallContext context) throws A2AClientException;
 
     /**
@@ -130,8 +133,9 @@ public abstract void sendMessage(Message request,
      * @param metadata the optional metadata to include when sending the message
      * @throws A2AClientException if sending the message fails for any reason
      */
-    public void sendMessage(Message request, PushNotificationConfig pushNotificationConfiguration,
-                            Map<String, Object> metadata) throws A2AClientException {
+    public void sendMessage(@NonNull Message request,
+                            @Nullable PushNotificationConfig pushNotificationConfiguration,
+                            @Nullable Map<String, Object> metadata) throws A2AClientException {
         sendMessage(request, pushNotificationConfiguration, metadata, null);
     }
 
@@ -147,11 +151,13 @@ public void sendMessage(Message request, PushNotificationConfig pushNotification
      * @param pushNotificationConfiguration the push notification configuration that should be
      *                                      used if the streaming approach is used
      * @param metadata the optional metadata to include when sending the message
-     * @param context optional client call context for the request (may be {@code null})
+     * @param context optional client call context for the request
      * @throws A2AClientException if sending the message fails for any reason
      */
-    public abstract void sendMessage(Message request, PushNotificationConfig pushNotificationConfiguration,
-                                     Map<String, Object> metadata, @Nullable ClientCallContext context) throws A2AClientException;
+    public abstract void sendMessage(@NonNull Message request,
+                                     @Nullable PushNotificationConfig pushNotificationConfiguration,
+                                     @Nullable Map<String, Object> metadata,
+                                     @Nullable ClientCallContext context) throws A2AClientException;
 
     /**
      * Retrieve the current state and history of a specific task.
@@ -320,8 +326,8 @@ public abstract void deleteTaskPushNotificationConfigurations(
      * @param request the parameters specifying which task's notification configs to delete
      * @throws A2AClientException if resubscribing fails for any reason
      */
-    public void resubscribe(TaskIdParams request) throws A2AClientException {
-        resubscribe(request, null);
+    public void resubscribe(@NonNull TaskIdParams request) throws A2AClientException {
+        resubscribe(request, consumers, streamingErrorHandler, null);
     }
 
     /**
@@ -332,10 +338,13 @@ public void resubscribe(TaskIdParams request) throws A2AClientException {
      * error handler will be used if an error occurs during streaming.
      *
      * @param request the parameters specifying which task's notification configs to delete
-     * @param context optional client call context for the request (may be {@code null})
+     * @param context optional client call context for the request
      * @throws A2AClientException if resubscribing fails for any reason
      */
-    public abstract void resubscribe(TaskIdParams request, @Nullable ClientCallContext context) throws A2AClientException;
+    public void resubscribe(@NonNull TaskIdParams request,
+                            @Nullable ClientCallContext context) throws A2AClientException {
+        resubscribe(request, consumers, streamingErrorHandler, context);
+    }
 
     /**
      * Resubscribe to a task's event stream.
@@ -349,8 +358,9 @@ public void resubscribe(TaskIdParams request) throws A2AClientException {
      * @param streamingErrorHandler an error handler that should be used for the streaming case if an error occurs
      * @throws A2AClientException if resubscribing fails for any reason
      */
-    public void resubscribe(TaskIdParams request, List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                            Consumer<Throwable> streamingErrorHandler) throws A2AClientException {
+    public void resubscribe(@NonNull TaskIdParams request,
+                            @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                            @Nullable Consumer<Throwable> streamingErrorHandler) throws A2AClientException {
         resubscribe(request, consumers, streamingErrorHandler, null);
     }
 
@@ -364,11 +374,13 @@ public void resubscribe(TaskIdParams request, List<BiConsumer<ClientEvent, Agent
      * @param request the parameters specifying which task's notification configs to delete
      * @param consumers a list of consumers to pass responses from the remote agent to
      * @param streamingErrorHandler an error handler that should be used for the streaming case if an error occurs
-     * @param context optional client call context for the request (may be {@code null})
+     * @param context optional client call context for the request
      * @throws A2AClientException if resubscribing fails for any reason
      */
-    public abstract void resubscribe(TaskIdParams request, List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                                     Consumer<Throwable> streamingErrorHandler, @Nullable ClientCallContext context) throws A2AClientException;
+    public abstract void resubscribe(@NonNull TaskIdParams request,
+                                     @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                                     @Nullable Consumer<Throwable> streamingErrorHandler,
+                                     @Nullable ClientCallContext context) throws A2AClientException;
 
     /**
      * Retrieve the AgentCard.
@@ -394,15 +406,6 @@ public AgentCard getAgentCard() throws A2AClientException {
      */
     public abstract void close();
 
-    /**
-     * Process the event using all configured consumers.
-     */
-    void consume(ClientEvent clientEventOrMessage, AgentCard agentCard) {
-        for (BiConsumer<ClientEvent, AgentCard> consumer : consumers) {
-            consumer.accept(clientEventOrMessage, agentCard);
-        }
-    }
-
     /**
      * Get the error handler that should be used during streaming.
      *

client/base/src/main/java/io/a2a/client/Client.java

@@ -56,30 +56,28 @@ public static ClientBuilder builder(AgentCard agentCard) {
     }
 
     @Override
-    public void sendMessage(Message request, @Nullable ClientCallContext context) throws A2AClientException {
-        MessageSendParams messageSendParams = getMessageSendParams(request, clientConfig);
-        sendMessage(messageSendParams, null, null, context);
-    }
-
-    @Override
-    public void sendMessage(Message request, List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                            Consumer<Throwable> streamingErrorHandler, @Nullable ClientCallContext context) throws A2AClientException {
+    public void sendMessage(@NonNull Message request,
+                            @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                            @Nullable Consumer<Throwable> streamingErrorHandler,
+                            @Nullable ClientCallContext context) throws A2AClientException {
         MessageSendParams messageSendParams = getMessageSendParams(request, clientConfig);
         sendMessage(messageSendParams, consumers, streamingErrorHandler, context);
     }
 
     @Override
-    public void sendMessage(Message request, PushNotificationConfig pushNotificationConfiguration,
-                            Map<String, Object> metatadata, @Nullable ClientCallContext context) throws A2AClientException {
+    public void sendMessage(@NonNull Message request,
+                            @Nullable PushNotificationConfig pushNotificationConfiguration,
+                            @Nullable Map<String, Object> metadata,
+                            @Nullable ClientCallContext context) throws A2AClientException {
         MessageSendConfiguration messageSendConfiguration = createMessageSendConfiguration(pushNotificationConfiguration);
 
         MessageSendParams messageSendParams = new MessageSendParams.Builder()
                 .message(request)
                 .configuration(messageSendConfiguration)
-                .metadata(metatadata)
+                .metadata(metadata)
                 .build();
 
-        sendMessage(messageSendParams, null, null, context);
+        sendMessage(messageSendParams, consumers, streamingErrorHandler, context);
     }
 
     @Override
@@ -122,14 +120,24 @@ public void deleteTaskPushNotificationConfigurations(
     }
 
     @Override
-    public void resubscribe(TaskIdParams request, @Nullable ClientCallContext context) throws A2AClientException {
-       resubscribeToTask(request, null, null, context);
-    }
-
-    @Override
-    public void resubscribe(TaskIdParams request, @Nullable List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                            @Nullable Consumer<Throwable> streamingErrorHandler, @Nullable ClientCallContext context) throws A2AClientException {
-        resubscribeToTask(request, consumers, streamingErrorHandler, context);
+    public void resubscribe(@NonNull TaskIdParams request,
+                            @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
+                            @Nullable Consumer<Throwable> streamingErrorHandler,
+                            @Nullable ClientCallContext context) throws A2AClientException {
+        if (! clientConfig.isStreaming() || ! agentCard.capabilities().streaming()) {
+            throw new A2AClientException("Client and/or server does not support resubscription");
+        }
+        ClientTaskManager tracker = new ClientTaskManager();
+        Consumer<Throwable> overriddenErrorHandler = getOverriddenErrorHandler(streamingErrorHandler);
+        Consumer<StreamingEventKind> eventHandler = event -> {
+            try {
+                ClientEvent clientEvent = getClientEvent(event, tracker);
+                consume(clientEvent, agentCard, consumers);
+            } catch (A2AClientError e) {
+                overriddenErrorHandler.accept(e);
+            }
+        };
+        clientTransport.resubscribe(request, eventHandler, overriddenErrorHandler, context);
     }
 
     @Override
@@ -169,7 +177,7 @@ private MessageSendConfiguration createMessageSendConfiguration(@Nullable PushNo
                 .build();
     }
 
-    private void sendMessage(MessageSendParams messageSendParams, @Nullable List<BiConsumer<ClientEvent, AgentCard>> consumers,
+    private void sendMessage(@NonNull MessageSendParams messageSendParams, @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers,
                              @Nullable Consumer<Throwable> errorHandler, @Nullable ClientCallContext context) throws A2AClientException {
         if (! clientConfig.isStreaming() || ! agentCard.capabilities().streaming()) {
             EventKind eventKind = clientTransport.sendMessage(messageSendParams, context);
@@ -196,24 +204,6 @@ private void sendMessage(MessageSendParams messageSendParams, @Nullable List<BiC
         }
     }
 
-    private void resubscribeToTask(TaskIdParams request, @Nullable List<BiConsumer<ClientEvent, AgentCard>> consumers,
-                                   @Nullable Consumer<Throwable> errorHandler, @Nullable ClientCallContext context) throws A2AClientException {
-        if (! clientConfig.isStreaming() || ! agentCard.capabilities().streaming()) {
-            throw new A2AClientException("Client and/or server does not support resubscription");
-        }
-        ClientTaskManager tracker = new ClientTaskManager();
-        Consumer<Throwable> overriddenErrorHandler = getOverriddenErrorHandler(errorHandler);
-        Consumer<StreamingEventKind> eventHandler = event -> {
-            try {
-                ClientEvent clientEvent = getClientEvent(event, tracker);
-                consume(clientEvent, agentCard, consumers);
-            } catch (A2AClientError e) {
-                overriddenErrorHandler.accept(e);
-            }
-        };
-        clientTransport.resubscribe(request, eventHandler, overriddenErrorHandler, context);
-    }
-
     private @NonNull Consumer<Throwable> getOverriddenErrorHandler(@Nullable Consumer<Throwable> errorHandler) {
         return e -> {
             if (errorHandler != null) {
@@ -226,15 +216,9 @@ private void resubscribeToTask(TaskIdParams request, @Nullable List<BiConsumer<C
         };
     }
 
-    private void consume(ClientEvent clientEvent, AgentCard agentCard, @Nullable List<BiConsumer<ClientEvent, AgentCard>> consumers) {
-        if (consumers != null) {
-            // use specified consumers
-            for (BiConsumer<ClientEvent, AgentCard> consumer : consumers) {
-                consumer.accept(clientEvent, agentCard);
-            }
-        } else {
-            // use configured consumers
-            consume(clientEvent, agentCard);
+    private void consume(ClientEvent clientEvent, AgentCard agentCard, @NonNull List<BiConsumer<ClientEvent, AgentCard>> consumers) {
+        for (BiConsumer<ClientEvent, AgentCard> consumer : consumers) {
+            consumer.accept(clientEvent, agentCard);
         }
     }