fix: Ensure proper errors are reported when authentication fails

Author: fjuma

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

@@ -7,15 +7,23 @@
 import java.net.http.HttpResponse;
 import java.net.http.HttpResponse.BodyHandler;
 import java.net.http.HttpResponse.BodyHandlers;
+import java.net.http.HttpResponse.BodySubscribers;
+import java.nio.ByteBuffer;
 import java.nio.charset.StandardCharsets;
 import java.util.HashMap;
+import java.util.List;
 import java.util.Map;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.Flow;
 import java.util.function.Consumer;
 
 public class JdkA2AHttpClient implements A2AHttpClient {
 
+    private static final int HTTP_OK = 200;
+    private static final int HTTP_MULTIPLE_CHOICES = 300;
+    private static final int HTTP_UNAUTHORIZED = 401;
+    private static final int HTTP_FORBIDDEN = 403;
+
     private final HttpClient httpClient;
 
     public JdkA2AHttpClient() {
@@ -88,6 +96,7 @@ protected CompletableFuture<Void> asyncRequest(
         ) {
             Flow.Subscriber<String> subscriber = new Flow.Subscriber<String>() {
                 private Flow.Subscription subscription;
+                private volatile boolean errorRaised = false;
 
                 @Override
                 public void onSubscribe(Flow.Subscription subscription) {
@@ -109,25 +118,75 @@ public void onNext(String item) {
 
                 @Override
                 public void onError(Throwable throwable) {
-                    errorConsumer.accept(throwable);
-                    subscription.cancel();
+                    if (!errorRaised) {
+                        errorRaised = true;
+                        errorConsumer.accept(throwable);
+                    }
+                    if (subscription != null) {
+                        subscription.cancel();
+                    }
                 }
 
                 @Override
                 public void onComplete() {
-                    completeRunnable.run();
-                    subscription.cancel();
+                    if (!errorRaised) {
+                        completeRunnable.run();
+                    }
+                    if (subscription != null) {
+                        subscription.cancel();
+                    }
                 }
             };
 
-            BodyHandler<Void> bodyHandler = BodyHandlers.fromLineSubscriber(subscriber);
+            // Create a custom body handler that checks status before processing body
+            BodyHandler<Void> bodyHandler = responseInfo -> {
+                // Check status code first, before creating the subscriber
+                if (!isSuccessStatus(responseInfo.statusCode())) {
+                    final String errorMessage;
+                    if (responseInfo.statusCode() == HTTP_UNAUTHORIZED) {
+                        errorMessage = "Authentication failed: Client credentials are missing or invalid";
+                    } else if (responseInfo.statusCode() == HTTP_FORBIDDEN) {
+                        errorMessage = "Authorization failed: Client does not have permission for the operation";
+                    } else {
+                        errorMessage = "Request failed with status " + responseInfo.statusCode();
+                    }
+                    // Return a body subscriber that immediately signals error
+                    return BodySubscribers.fromSubscriber(new Flow.Subscriber<List<ByteBuffer>>() {
+                        @Override
+                        public void onSubscribe(Flow.Subscription subscription) {
+                            subscriber.onError(new IOException(errorMessage));
+                        }
+                        
+                        @Override
+                        public void onNext(List<ByteBuffer> item) {
+                            // Should not be called
+                        }
+                        
+                        @Override
+                        public void onError(Throwable throwable) {
+                            // Should not be called
+                        }
+                        
+                        @Override
+                        public void onComplete() {
+                            // Should not be called
+                        }
+                    });
+                } else {
+                    // Status is OK, proceed with normal line subscriber
+                    return BodyHandlers.fromLineSubscriber(subscriber).apply(responseInfo);
+                }
+            };
 
             // Send the response async, and let the subscriber handle the lines.
             return httpClient.sendAsync(request, bodyHandler)
                     .thenAccept(response -> {
-                        if (!JdkHttpResponse.success(response)) {
-                            subscriber.onError(new IOException("Request failed " + response.statusCode()));
-                        }
+                        // Status checking is now handled in the body handler
+                    })
+                    .exceptionally(throwable -> {
+                        // handle any other async errors (network issues, etc.)
+                        subscriber.onError(new IOException("Request failed: " + throwable.getMessage(), throwable));
+                        return null;
                     });
         }
     }
@@ -200,6 +259,13 @@ public A2AHttpResponse post() throws IOException, InterruptedException {
                     .build();
             HttpResponse<String> response =
                     httpClient.send(request, BodyHandlers.ofString(StandardCharsets.UTF_8));
+
+            if (response.statusCode() == HTTP_UNAUTHORIZED) {
+                throw new IOException("Authentication failed: Client credentials are missing or invalid");
+            } else if (response.statusCode() == HTTP_FORBIDDEN) {
+                throw new IOException("Authorization failed: Client does not have permission for the operation");
+            }
+            
             return new JdkHttpResponse(response);
         }
 
@@ -227,12 +293,16 @@ public boolean success() {// Send the request and get the response
         }
 
         static boolean success(HttpResponse<?> response) {
-            return response.statusCode() >= 200 && response.statusCode() < 300;
+            return response.statusCode() >= HTTP_OK && response.statusCode() < HTTP_MULTIPLE_CHOICES;
         }
 
         @Override
         public String body() {
             return response.body();
         }
     }
+
+    private static boolean isSuccessStatus(int statusCode) {
+        return statusCode >= HTTP_OK && statusCode < HTTP_MULTIPLE_CHOICES;
+    }
 }