diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java index 29dc23c35..5044e8f75 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/client/transport/ResponseSubscribers.java @@ -149,6 +149,11 @@ protected void hookOnNext(String line) { this.sink.next(new SseResponseEvent(responseInfo, sseEvent)); this.eventBuilder.setLength(0); + // Per the WHATWG SSE spec, both the data buffer and the event type + // buffer are reset when an event is dispatched, so a stale event type + // must not leak into the following events. + this.currentEventId.set(null); + this.currentEventType.set(null); } } else { @@ -194,6 +199,9 @@ protected void hookOnComplete() { String eventData = this.eventBuilder.toString(); SseEvent sseEvent = new SseEvent(currentEventId.get(), currentEventType.get(), eventData.trim()); this.sink.next(new SseResponseEvent(responseInfo, sseEvent)); + this.eventBuilder.setLength(0); + this.currentEventId.set(null); + this.currentEventType.set(null); } this.sink.complete(); } diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTests.java b/mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTests.java new file mode 100644 index 000000000..f8ae07a3f --- /dev/null +++ b/mcp-core/src/test/java/io/modelcontextprotocol/client/transport/ResponseSubscribersTests.java @@ -0,0 +1,76 @@ +/* + * Copyright 2024-2026 the original author or authors. + */ + +package io.modelcontextprotocol.client.transport; + +import java.net.http.HttpResponse.ResponseInfo; +import java.util.List; + +import org.junit.jupiter.api.Test; +import org.reactivestreams.Subscription; + +import io.modelcontextprotocol.client.transport.ResponseSubscribers.ResponseEvent; +import io.modelcontextprotocol.client.transport.ResponseSubscribers.SseLineSubscriber; +import io.modelcontextprotocol.client.transport.ResponseSubscribers.SseResponseEvent; +import reactor.core.publisher.Flux; + +import static org.assertj.core.api.Assertions.assertThat; +import static org.mockito.Mockito.mock; + +/** + * Unit tests for {@link ResponseSubscribers.SseLineSubscriber} event buffer handling. + * + *
+ * Per the WHATWG
+ * HTML Living Standard ยง9.2.6, both the data buffer and the event type buffer must be
+ * reset when an event is dispatched, so that a stale event type or id does not leak into
+ * subsequent events.
+ */
+class ResponseSubscribersTests {
+
+ @Test
+ void eventTypeIsResetAfterEventDispatch() {
+ List