fix: propagate SSE stream errors to messageEndpointSink

When the SSE stream encounters an error before the endpoint event is
received, the messageEndpointSink is never signaled. This causes
sendMessage() to hang indefinitely since Sinks.One.asMono() has no
built-in timeout.

This commit propagates errors to messageEndpointSink in two places:

1. onErrorComplete handler: propagates the actual error
2. doFinally handler: emits CancellationException as a safety net for
   cases where the stream completes without going through onErrorComplete

Since Sinks.One only accepts a single emission, duplicate calls to
tryEmitError() are safely ignored when the sink has already been
completed (either with a value or an error).

Closes #323

Signed-off-by: gyeo009 <gyeo009@users.noreply.github.com>

Author: gyeo009

mcp-core/src/main/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransport.java

@@ -11,6 +11,7 @@
 import java.net.http.HttpResponse;
 import java.time.Duration;
 import java.util.List;
+import java.util.concurrent.CancellationException;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.atomic.AtomicReference;
 import java.util.function.Consumer;
@@ -399,9 +400,13 @@ else if (MESSAGE_EVENT_TYPE.equals(responseEvent.sseEvent().event())) {
 						logger.warn("SSE stream observed an error", t);
 						sink.error(t);
 					}
+					this.messageEndpointSink.tryEmitError(t);
 					return true;
 				})
 				.doFinally(s -> {
+					this.messageEndpointSink
+						.tryEmitError(new CancellationException("SSE stream ended before receiving the endpoint event"));
+
 					Disposable ref = this.sseSubscription.getAndSet(null);
 					if (ref != null && !ref.isDisposed()) {
 						ref.dispose();

mcp-test/src/test/java/io/modelcontextprotocol/client/transport/HttpClientSseClientTransportTests.java

@@ -477,4 +477,21 @@ void testAsyncRequestCustomizer() {
 		customizedTransport.closeGracefully().block();
 	}
 
+	@Test
+	void testSendMessageFailsWhenSseConnectionFails() {
+		HttpClientSseClientTransport failingTransport = HttpClientSseClientTransport.builder("http://non-existent-host")
+			.connectTimeout(Duration.ofSeconds(2))
+			.build();
+
+		failingTransport.connect(Function.identity()).subscribe();
+
+		JSONRPCRequest testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
+				Map.of("key", "value"));
+
+		// sendMessage() should fail with an error rather than hanging forever
+		StepVerifier.create(failingTransport.sendMessage(testMessage)).expectError().verify(Duration.ofSeconds(10));
+
+		failingTransport.closeGracefully().block();
+	}
+
 }