Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -13,7 +13,6 @@
import java.util.concurrent.CompletionStage;
import java.util.concurrent.Flow;
import java.util.concurrent.atomic.AtomicReference;
import java.util.regex.Pattern;

import org.reactivestreams.FlowAdapters;
import org.reactivestreams.Subscription;
Expand Down Expand Up @@ -147,21 +146,6 @@ static BodyHandler<String> boundedStringBodyHandler(int maxSize) {

static class SseLineSubscriber extends BaseSubscriber<String> {

/**
* Pattern to extract data content from SSE "data:" lines.
*/
private static final Pattern EVENT_DATA_PATTERN = Pattern.compile("^data:(.+)$", Pattern.MULTILINE);

/**
* Pattern to extract event ID from SSE "id:" lines.
*/
private static final Pattern EVENT_ID_PATTERN = Pattern.compile("^id:(.+)$", Pattern.MULTILINE);

/**
* Pattern to extract event type from SSE "event:" lines.
*/
private static final Pattern EVENT_TYPE_PATTERN = Pattern.compile("^event:(.+)$", Pattern.MULTILINE);

/**
* The sink for emitting parsed response events.
*/
Expand Down Expand Up @@ -227,6 +211,31 @@ protected void hookOnSubscribe(Subscription subscription) {
});
}

/**
* Extracts the value of an SSE field from a line, per the <a href=
* "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">
* SSE specification</a>: the characters after the colon with a single leading
* space removed.
*
* <p>
* A value may legally contain U+2028 (LINE SEPARATOR), U+2029 (PARAGRAPH
* SEPARATOR) and U+0085 (NEXT LINE). Those are not SSE line terminators, so
* extracting the value with a {@code MULTILINE} regex instead of this method
* silently truncates it there.
* @param line the SSE line, already stripped of its terminator by the line
* subscriber
* @param field the field prefix, e.g. {@code "data:"}
* @return the field value with a single leading space removed, never truncated
* @see #hookOnNext(String)
*/
private static String fieldValue(String line, String field) {
String value = line.substring(field.length());
if (value.startsWith(" ")) {
value = value.substring(1);
}
return value;
}

@Override
protected void hookOnNext(String line) {
if (line.isEmpty()) {
Expand All @@ -241,35 +250,25 @@ protected void hookOnNext(String line) {
}
else {
if (line.startsWith("data:")) {
var matcher = EVENT_DATA_PATTERN.matcher(line);
if (matcher.find()) {
String data = matcher.group(1).trim();
// Measured before appending, so that an event carrying exactly
// maxSize of data is accepted: the trailing separator below is
// stripped again before the event is emitted.
if (this.eventBuilder.length() + data.length() > this.maxSize) {
upstream().cancel();
this.sink.error(
new McpTransportException("Inbound SSE event exceeds the maximum allowed size of "
+ this.maxSize + " bytes"));
return;
}
this.eventBuilder.append(data).append("\n");
String data = fieldValue(line, "data:");
// Measured before appending, so that an event carrying exactly
// maxSize of data is accepted: the trailing separator below is
// stripped again before the event is emitted.
if (this.eventBuilder.length() + data.length() > this.maxSize) {
upstream().cancel();
this.sink.error(new McpTransportException(
"Inbound SSE event exceeds the maximum allowed size of " + this.maxSize + " bytes"));
return;
}
this.eventBuilder.append(data).append("\n");
upstream().request(1);
}
else if (line.startsWith("id:")) {
var matcher = EVENT_ID_PATTERN.matcher(line);
if (matcher.find()) {
this.currentEventId.set(matcher.group(1).trim());
}
this.currentEventId.set(fieldValue(line, "id:"));
upstream().request(1);
}
else if (line.startsWith("event:")) {
var matcher = EVENT_TYPE_PATTERN.matcher(line);
if (matcher.find()) {
this.currentEventType.set(matcher.group(1).trim());
}
this.currentEventType.set(fieldValue(line, "event:"));
upstream().request(1);
}
else if (line.startsWith(":")) {
Expand Down
Original file line number Diff line number Diff line change
@@ -0,0 +1,120 @@
/*
* Copyright 2024-2026 the original author or authors.
*/

package io.modelcontextprotocol.client.transport;

import java.net.http.HttpClient;
import java.net.http.HttpHeaders;
import java.net.http.HttpResponse;
import java.util.List;
import java.util.Map;

import org.junit.jupiter.api.Test;

import reactor.core.publisher.Flux;
import reactor.test.StepVerifier;

import static org.assertj.core.api.Assertions.assertThat;

/**
* Unit tests for {@link ResponseSubscribers.SseLineSubscriber}.
*
* <p>
* Verifies that SSE field values are extracted per the <a href=
* "https://html.spec.whatwg.org/multipage/server-sent-events.html#event-stream-interpretation">
* WHATWG HTML Living Standard §9.2.6</a>: the field value is everything after the colon
* minus a single leading space. In particular, U+2028 (LINE SEPARATOR), U+2029 (PARAGRAPH
* SEPARATOR) and U+0085 (NEXT LINE) are legal inside a field value and must not truncate
* it — they are not SSE line terminators.
*
* @see <a href="https://github.com/modelcontextprotocol/java-sdk/issues/1136">#1136</a>
*/
class ResponseSubscribersTest {

private static final HttpResponse.ResponseInfo RESPONSE_INFO = new HttpResponse.ResponseInfo() {

@Override
public int statusCode() {
return 200;
}

@Override
public HttpHeaders headers() {
return HttpHeaders.of(Map.of(), (name, value) -> true);
}

@Override
public HttpClient.Version version() {
return HttpClient.Version.HTTP_1_1;
}

};

private static List<ResponseSubscribers.SseEvent> parse(List<String> lines) {
return Flux.<ResponseSubscribers.ResponseEvent>create(sink -> Flux.fromIterable(lines)
.subscribe(new ResponseSubscribers.SseLineSubscriber(RESPONSE_INFO, sink, Integer.MAX_VALUE)))
.map(event -> ((ResponseSubscribers.SseResponseEvent) event).sseEvent())
.collectList()
.block();
}

/**
* A {@code data:} payload containing U+2028, U+2029 or U+0085 must survive parsing
* intact. A MULTILINE regex used to truncate the value at those characters, because
* the Java regex engine treats them as line terminators.
*/
@Test
void shouldNotTruncateDataAtUnicodeLineSeparators() {
List<String> separators = List.of("\u2028", "\u2029", "\u0085");

for (String separator : separators) {
String payload = "{\"text\":\"a" + separator + "b\"}";
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: " + payload, ""));

assertThat(events).as("payload with U+%04X", (int) separator.charAt(0)).hasSize(1);
assertThat(events.get(0).data()).isEqualTo(payload);
}
}

@Test
void shouldPreserveVerticalTabInData() {
String payload = "{\"text\":\"a\u000Bb\"}";

List<ResponseSubscribers.SseEvent> events = parse(List.of("data: " + payload, ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo(payload);
}

@Test
void shouldStripOnlySingleLeadingSpacePerDataLine() {
// The leading space after the colon is stripped per line; any further
// whitespace is part of the value. The whole-event trim only affects the
// first/last line, so the interior line keeps its second space.
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data: second", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo("first\n second");
}

@Test
void shouldJoinMultipleDataLinesWithNewline() {
List<ResponseSubscribers.SseEvent> events = parse(List.of("data: first", "data: second", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).data()).isEqualTo("first\nsecond");
}

@Test
void shouldCaptureEventIdAndTypeWithUnicodeValue() {
List<ResponseSubscribers.SseEvent> events = parse(
List.of("event: message\u2028tail", "id: 42\u2028tail", "data: body", ""));

assertThat(events).hasSize(1);
assertThat(events.get(0).event()).isEqualTo("message\u2028tail");
assertThat(events.get(0).id()).isEqualTo("42\u2028tail");
assertThat(events.get(0).data()).isEqualTo("body");
}

}