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 completed without 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,29 @@ void testAsyncRequestCustomizer() {
 		customizedTransport.closeGracefully().block();
 	}
 
+	@Test
+	void testSendMessageFailsWhenSseConnectionFails() {
+		// Create a transport that connects to a non-existent host.
+		// The SSE connection will fail, and sendMessage() should propagate
+		// the error instead of hanging indefinitely.
+		// See: https://github.com/modelcontextprotocol/java-sdk/pull/323
+		HttpClientSseClientTransport failingTransport = HttpClientSseClientTransport.builder("http://non-existent-host")
+			.connectTimeout(Duration.ofSeconds(2))
+			.build();
+
+		// Attempt to connect - this will fail due to the non-existent host
+		failingTransport.connect(Function.identity()).subscribe();
+
+		// sendMessage() should fail with an error rather than hanging forever.
+		// Without the fix, messageEndpointSink never receives a signal and
+		// asMono() blocks indefinitely, causing this test to time out.
+		JSONRPCRequest testMessage = new JSONRPCRequest(McpSchema.JSONRPC_VERSION, "test-method", "test-id",
+				Map.of("key", "value"));
+
+		StepVerifier.create(failingTransport.sendMessage(testMessage)).expectError().verify(Duration.ofSeconds(10));
+
+		// Clean up
+		failingTransport.closeGracefully().block();
+	}
+
 }