Polish PR

Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>

Author: Kehrlann

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

@@ -393,7 +393,10 @@ public Mono<Void> connect(Function<Mono<JSONRPCMessage>, Mono<JSONRPCMessage>> h
 						ResponseSubscribers.boundedPublisherBodyHandler(this.maxResponseSize)))
 				.flatMapMany(response -> {
 					if (isClosing) {
-						return Flux.empty();
+						// The body is handed over as a publisher and nothing is read off
+						// the wire until it is subscribed, so it has to be drained even
+						// when its content is of no further interest.
+						return ResponseSubscribers.drain(response.body());
 					}
 
 					int statusCode = response.statusCode();

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

@@ -286,6 +286,9 @@ private Flux<McpSchema.JSONRPCMessage> consumeSseStream(
 		return ResponseSubscribers.decodeSseResponse(lines, this.maxResponseSize).flatMap(sseEvent -> {
 			if (!isMessageEvent(sseEvent.event())) {
 				logger.debug("Received SSE event with type: {}", sseEvent);
+				if (onFirstMessage != null) {
+					onFirstMessage.run();
+				}
 				return Flux.empty();
 			}
 			String data = sseEvent.data();
@@ -536,8 +539,6 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 							ResponseSubscribers.boundedPublisherBodyHandler(this.maxResponseSize)))
 					.flatMapMany(httpResponse -> {
 						int statusCode = httpResponse.statusCode();
-						Exception exception = null;
-						boolean proceed = false;
 						Optional<String> maybeSessionId = transportSession == null ? Optional.empty()
 								: transportSession.sessionId();
 						if (statusCode == 401 || statusCode == 403) {
@@ -552,8 +553,7 @@ public Mono<Void> sendMessage(McpSchema.JSONRPCMessage sentMessage) {
 						}
 
 						if (transportSession
-							.markInitialized(httpResponse.headers().firstValue("mcp-session-id").orElse(null))
-								&& !openConnectionOnStartup) {
+							.markInitialized(httpResponse.headers().firstValue("mcp-session-id").orElse(null))) {
 							reconnect(null).contextWrite(deliveredSink.contextView()).subscribe();
 						}
 

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

@@ -12,6 +12,7 @@
 import java.nio.charset.CharacterCodingException;
 import java.nio.charset.CharsetDecoder;
 import java.nio.charset.CoderResult;
+import java.nio.charset.CodingErrorAction;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
 import java.util.List;
@@ -180,12 +181,25 @@ static <T> Flux<T> drain(Publisher<List<ByteBuffer>> body) {
 
 	/**
 	 * Stateful UTF-8 decoder that splits a stream of byte-buffer chunks into complete
-	 * lines. Handles multi-byte characters split across chunk boundaries, and both
-	 * {@code "\n"} and {@code "\r\n"} terminators.
+	 * lines. Handles multi-byte characters split across chunk boundaries, and terminates
+	 * a line on {@code "\r\n"}, {@code "\r"} or {@code "\n"} alike, as the SSE wire
+	 * format does. Bytes that do not decode are replaced rather than reported, so a peer
+	 * sending one does not cost the stream.
 	 */
 	static final class Utf8LineDecoder {
 
-		private final CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder();
+		/**
+		 * Undecodable input costs one replacement character rather than the stream: a
+		 * decoder left on the default {@link CodingErrorAction#REPORT} fails the whole
+		 * response over a single byte a peer mangled, and takes with it the lines already
+		 * decoded from the same chunk, because {@link #decode(List)} throws instead of
+		 * returning them. A body cut short mid-character is enough to hit it. This
+		 * matches {@link java.net.http.HttpResponse.BodySubscribers#fromLineSubscriber},
+		 * the path this decoder replaces, which configured the same two actions.
+		 */
+		private final CharsetDecoder decoder = StandardCharsets.UTF_8.newDecoder()
+			.onMalformedInput(CodingErrorAction.REPLACE)
+			.onUnmappableCharacter(CodingErrorAction.REPLACE);
 
 		private final CharBuffer charBuffer = CharBuffer.allocate(4096);
 
@@ -202,6 +216,14 @@ static final class Utf8LineDecoder {
 		 */
 		private int scannedForLineTerminator = 0;
 
+		/**
+		 * Whether the line just emitted was terminated by a CR, so that a LF opening what
+		 * follows completes that terminator instead of ending a line of its own. A CR is
+		 * emitted on as soon as it arrives, before it is known whether a LF follows it,
+		 * and the two may be split across chunks.
+		 */
+		private boolean crTerminatedPreviousLine = false;
+
 		// Holds partial UTF-8 sequences left over from a previous chunk (max 3 bytes
 		// for a BMP code point; 4 bytes for a supplementary one).
 		private ByteBuffer pendingBytes = ByteBuffer.allocate(0);
@@ -221,6 +243,9 @@ List<String> decode(List<ByteBuffer> chunk) {
 					CoderResult result = decoder.decode(input, charBuffer, false);
 					drainCharBuffer();
 					extractCompletedLines(lines);
+					// Unreachable while the decoder replaces undecodable input, but kept
+					// so that an error result cannot spin this loop: it is neither an
+					// underflow nor an overflow.
 					if (result.isError()) {
 						try {
 							result.throwException();
@@ -270,13 +295,11 @@ List<String> flush() {
 			extractCompletedLines(lines);
 			if (leftover.length() > 0) {
 				String last = leftover.toString();
-				if (last.endsWith("\r")) {
-					last = last.substring(0, last.length() - 1);
-				}
 				leftover.setLength(0);
 				this.scannedForLineTerminator = 0;
 				lines.add(last);
 			}
+			this.crTerminatedPreviousLine = false;
 			return lines;
 		}
 
@@ -287,19 +310,43 @@ private void drainCharBuffer() {
 		}
 
 		private void extractCompletedLines(List<String> out) {
-			int newlineIdx;
-			while ((newlineIdx = leftover.indexOf("\n", this.scannedForLineTerminator)) != -1) {
-				String line = leftover.substring(0, newlineIdx);
-				if (line.endsWith("\r")) {
-					line = line.substring(0, line.length() - 1);
+			while (true) {
+				if (this.crTerminatedPreviousLine) {
+					if (leftover.length() == 0) {
+						// The LF, if there is one, is in a chunk that has not arrived.
+						return;
+					}
+					if (leftover.charAt(0) == '\n') {
+						leftover.delete(0, 1);
+					}
+					this.crTerminatedPreviousLine = false;
 				}
-				out.add(line);
-				leftover.delete(0, newlineIdx + 1);
+				int terminatorIdx = indexOfLineTerminator(this.scannedForLineTerminator);
+				if (terminatorIdx == -1) {
+					this.scannedForLineTerminator = leftover.length();
+					return;
+				}
+				out.add(leftover.substring(0, terminatorIdx));
+				this.crTerminatedPreviousLine = leftover.charAt(terminatorIdx) == '\r';
+				leftover.delete(0, terminatorIdx + 1);
 				// What is left starts after the terminator, so none of it has been
 				// searched yet.
 				this.scannedForLineTerminator = 0;
 			}
-			this.scannedForLineTerminator = leftover.length();
+		}
+
+		/**
+		 * Index of the first CR or LF in {@link #leftover} at or after {@code from}, or
+		 * {@code -1} when there is none.
+		 */
+		private int indexOfLineTerminator(int from) {
+			for (int i = from; i < leftover.length(); i++) {
+				char c = leftover.charAt(i);
+				if (c == '\n' || c == '\r') {
+					return i;
+				}
+			}
+			return -1;
 		}
 
 	}
@@ -479,7 +526,7 @@ public void onComplete() {
 
 	/**
 	 * A {@link BoundedBodySubscriber} that aborts the response once a single line (a run
-	 * of bytes with no LF) exceeds {@code maxSize} bytes.
+	 * of bytes with no line terminator) exceeds {@code maxSize} bytes.
 	 *
 	 * <p>
 	 * {@link Utf8LineDecoder} buffers characters until it encounters a line terminator,
@@ -488,9 +535,10 @@ public void onComplete() {
 	 * wire and cancels the subscription before that buffer can grow without bound.
 	 *
 	 * <p>
-	 * Only LF resets the count, because LF is the only byte {@link Utf8LineDecoder}
-	 * flushes a line on: a lone CR leaves the decoder's buffer growing, so it must not
-	 * refill this budget either. CRLF still resets, on its LF.
+	 * CR and LF both reset the count, matching the terminators {@link Utf8LineDecoder}
+	 * flushes a line on: whatever empties the decoder's buffer has to refill this budget,
+	 * or a peer framing short lines with CR alone would be aborted for exceeding a bound
+	 * its lines never reach. A CRLF resets twice, which is harmless.
 	 */
 	static final class BoundedLineBodySubscriber<T> extends BoundedBodySubscriber<T> {
 
@@ -518,7 +566,7 @@ protected boolean checkSize(ByteBuffer buffer) {
 			// The limit is within reach, so account for every line exactly.
 			for (int i = position; i < limit; i++) {
 				byte b = buffer.get(i);
-				if (b == '\n') {
+				if (b == '\n' || b == '\r') {
 					this.bytesSinceLineTerminator = 0;
 				}
 				else if (++this.bytesSinceLineTerminator > this.maxSize) {
@@ -529,12 +577,13 @@ else if (++this.bytesSinceLineTerminator > this.maxSize) {
 		}
 
 		/**
-		 * Returns the number of bytes after the last LF in the buffer, or the whole span
-		 * added to the running count when the buffer holds no LF.
+		 * Returns the number of bytes after the last line terminator in the buffer, or
+		 * the whole span added to the running count when the buffer holds none.
 		 */
 		private long lengthOfTrailingRun(ByteBuffer buffer, int position, int limit) {
 			for (int i = limit - 1; i >= position; i--) {
-				if (buffer.get(i) == '\n') {
+				byte b = buffer.get(i);
+				if (b == '\n' || b == '\r') {
 					return limit - 1 - i;
 				}
 			}

mcp-core/src/test/java/io/modelcontextprotocol/client/transport/BoundedBodySubscriberTests.java

@@ -107,17 +107,18 @@ void lineSubscriberHandlesCrLfSplitAcrossBuffers() {
 	}
 
 	@Test
-	void lineSubscriberIsNotResetByLoneCarriageReturns() {
+	void lineSubscriberIsResetByLoneCarriageReturns() {
 		BoundedLineBodySubscriber<Void> subscriber = lineSubscriber();
 
-		// Utf8LineDecoder only flushes a line on LF, so a peer streaming CR-terminated
-		// runs keeps its buffer growing. A lone CR must not refill the budget.
+		// A lone CR terminates a line, so it flushes Utf8LineDecoder's buffer and has to
+		// refill the budget here too. Otherwise a peer framing short lines with CR alone
+		// has its response aborted for exceeding a bound its lines never reach.
 		boolean accepted = true;
 		for (int i = 0; i < 10 && accepted; i++) {
 			accepted = subscriber.checkSize(buffer("aa\r"));
 		}
 
-		assertThat(accepted).isFalse();
+		assertThat(accepted).isTrue();
 	}
 
 	@Test

mcp-core/src/test/java/io/modelcontextprotocol/client/transport/HttpClientHttpTransportLeakTests.java

@@ -0,0 +1,107 @@
+/*
+ * Copyright 2026-2026 the original author or authors.
+ */
+
+package io.modelcontextprotocol.client.transport;
+
+import java.util.List;
+import java.util.Map;
+import java.util.function.Function;
+import java.util.stream.Stream;
+
+import io.modelcontextprotocol.spec.McpClientTransport;
+import io.modelcontextprotocol.spec.McpSchema;
+import io.modelcontextprotocol.spec.json.gson.GsonMcpJsonMapper;
+import org.junit.jupiter.params.ParameterizedTest;
+import org.junit.jupiter.params.provider.Arguments;
+import org.junit.jupiter.params.provider.MethodSource;
+import reactor.test.StepVerifier;
+
+import static org.assertj.core.api.Assertions.assertThat;
+import static org.junit.jupiter.api.Named.named;
+import static org.junit.jupiter.params.provider.Arguments.arguments;
+
+class HttpClientHttpTransportLeakTests {
+
+	static int selectorManagerThreadCount() {
+		return selectorManagerThreadNames().size();
+	}
+
+	static List<String> selectorManagerThreadNames() {
+		return Thread.getAllStackTraces()
+			.keySet()
+			.stream()
+			.map(Thread::getName)
+			.filter(name -> name.contains("HttpClient") && name.contains("SelectorManager"))
+			.sorted()
+			.toList();
+	}
+
+	static int forceGcUntilStable() throws InterruptedException {
+		int previousCount = Integer.MAX_VALUE;
+		int stableIterations = 0;
+		int currentCount = previousCount;
+
+		for (int i = 0; i < 40; i++) {
+			System.gc();
+			System.runFinalization();
+			Thread.sleep(250);
+
+			currentCount = selectorManagerThreadCount();
+			if (currentCount == previousCount) {
+				stableIterations++;
+				if (stableIterations >= 4) {
+					break;
+				}
+			}
+			else {
+				stableIterations = 0;
+				previousCount = currentCount;
+			}
+		}
+
+		return currentCount;
+	}
+
+	static void pauseForSelectorStartup() throws InterruptedException {
+		Thread.sleep(150);
+	}
+
+	@ParameterizedTest
+	@MethodSource("httpTransports")
+	void closeDoesNotRetainOwnedHttpClient(Function<String, McpClientTransport> httpTransportBuilder) throws Exception {
+		try (LoopbackMcpHttpServer server = LoopbackMcpHttpServer.start()) {
+			int selectorThreadsBefore = selectorManagerThreadCount();
+			Function<reactor.core.publisher.Mono<McpSchema.JSONRPCMessage>, reactor.core.publisher.Mono<McpSchema.JSONRPCMessage>> handler = Function
+				.identity();
+
+			for (int i = 0; i < 12; i++) {
+				McpClientTransport transport = httpTransportBuilder.apply(server.baseUri().toString());
+
+				StepVerifier.create(transport.connect(handler)).verifyComplete();
+				StepVerifier.create(transport.sendMessage(
+						new McpSchema.JSONRPCNotification(McpSchema.JSONRPC_VERSION, "ping", Map.of("iteration", i))))
+					.verifyComplete();
+				pauseForSelectorStartup();
+				StepVerifier.create(transport.closeGracefully()).verifyComplete();
+			}
+
+			int selectorThreadsAfter = forceGcUntilStable();
+
+			assertThat(selectorThreadsAfter)
+				.describedAs(
+						"closed transports should not keep owned HttpClient instances alive, remaining threads: %s",
+						selectorManagerThreadNames())
+				.isLessThanOrEqualTo(selectorThreadsBefore + 1);
+		}
+	}
+
+	static Stream<Arguments> httpTransports() {
+		Function<String, McpClientTransport> streamableHttp = (
+				uri) -> HttpClientStreamableHttpTransport.builder(uri).jsonMapper(new GsonMcpJsonMapper()).build();
+		Function<String, McpClientTransport> sse = (
+				uri) -> HttpClientSseClientTransport.builder(uri).jsonMapper(new GsonMcpJsonMapper()).build();
+		return Stream.of(arguments(named("Streamable HTTP", streamableHttp)), arguments(named("SSE", sse)));
+	}
+
+}