refactor: improve HTTP client error handling and cleanup debug code

- Replace exceptionallyCompose with whenComplete for better async error handling
- Make ResponseSubscribers class package-private

Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>

Author: tzolov

mcp/src/main/java/io/modelcontextprotocol/client/transport/HttpClientStreamableHttpTransport.java

@@ -162,10 +162,13 @@ private Publisher<Void> createDelete(String sessionId) {
 				.build();
 
 			return Mono.fromFuture(() -> this.httpClient.sendAsync(request, HttpResponse.BodyHandlers.ofString())
-				.exceptionallyCompose(e -> {
-					logger.warn("Error sending message", e);
-
-					return CompletableFuture.failedFuture(e);
+				.whenComplete((response, throwable) -> {
+					if (throwable != null) {
+						logger.warn("Error sending message", throwable);
+					}
+					else {
+						logger.debug("SSE connection established successfully");
+					}
 				})).doOnError(e -> logger.warn("Got error when closing transport", e)).then();
 		});
 	}
@@ -233,10 +236,14 @@ private Mono<Disposable> reconnect(McpTransportStream<Disposable> stream) {
 
 			Disposable connection = Flux.<ResponseEvent>create(sseSink -> this.httpClient
 				.sendAsync(request, responseInfo -> ResponseSubscribers.sseToBodySubscriber(responseInfo, sseSink))
-				.exceptionallyCompose(e -> {
-					logger.warn("Error sending message", e);
-					sseSink.error(e);
-					return CompletableFuture.failedFuture(e);
+				.whenComplete((response, throwable) -> {
+					if (throwable != null) {
+						logger.warn("Error sending message", throwable);
+						sseSink.error(throwable);
+					}
+					else {
+						logger.debug("SSE connection established successfully");
+					}
 				})).flatMap(responseEvent -> {
 					int statusCode = responseEvent.responseInfo().statusCode();
 
@@ -369,10 +376,14 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sendMessage) {
 
 				// Create the async request with proper body subscriber selection
 				Mono.fromFuture(this.httpClient.sendAsync(request, this.toSendMessageBodySubscriber(responseEventSink))
-					.exceptionallyCompose(e -> {
-						logger.warn("Error sending message", e);
-						responseEventSink.error(e);
-						return CompletableFuture.failedFuture(e);
+					.whenComplete((response, throwable) -> {
+						if (throwable != null) {
+							logger.warn("Error sending message", throwable);
+							responseEventSink.error(throwable);
+						}
+						else {
+							logger.debug("SSE connection established successfully");
+						}
 					})).subscribe();
 
 			}).flatMap(responseEvent -> {

mcp/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java

@@ -20,7 +20,7 @@
 import reactor.core.publisher.BaseSubscriber;
 import reactor.core.publisher.FluxSink;
 
-public class ResponseSubscribers {
+class ResponseSubscribers {
 
 	/**
 	 * Represents a Server-Sent Event with its standard fields.
@@ -328,7 +328,6 @@ protected void hookOnSubscribe(Subscription subscription) {
 
 		@Override
 		protected void hookOnNext(String line) {
-			System.out.println(">>>>>>>>>>>>>> Received line: " + line);
 		}
 
 		/**