FazBrowse GitHub Viewer | Trending |
URL:
| Home
Tools: [Download Repo ZIP]   [Original HTTPS Page]

fix: StdioServerTransportProvider readiness barrier uses wrong Mono operator by Debasishhh · Pull Request #1138 · modelcontextprotocol/java-sdk · GitHub

  1. modelcontextprotocol
  2. java-sdk

Repository navigation

fix: StdioServerTransportProvider readiness barrier uses wrong Mono operator #1138

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Debasishhh wants to merge 2 commits into
base: main
Choose a base branch
Could not load branches
Branch not found: {{ refName }}

Are you sure you want to change the base?

Some commits from the old base branch may be removed from the timeline, and old review comments may become outdated.
from
Conversation 0 Commits 2 Checks 0 Files changed
Open

fix: StdioServerTransportProvider readiness barrier uses wrong Mono operator #1138

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f076eb2
fix stdio transport readiness barrier using wrong Mono operator
Debasishhh Sep 20, 2026
b9375e2
add regression test for Mono.zip vs Mono.when readiness semantics
Debasishhh Sep 20, 2026
File filter

Filter by extension

Filter by extension .java  (2) All 1 file type selected
Viewed files
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Unified
Split
Hide whitespace
Diff view
Unified
Split
Hide whitespace
2 changes: 1 addition & 1 deletion .../src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -174,7 +174,7 @@ public StdioMcpSessionTransport() {
@Override
public Mono<Void> sendMessage(McpSchema.JSONRPCMessage message) {

return Mono.zip(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> {
return Mono.when(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> {
try {
outboundSink.emitNext(message, Sinks.EmitFailureHandler.busyLooping(Duration.ofMillis(100)));
return Mono.empty();
Expand Down
44 changes: 44 additions & 0 deletions ...test/java/io/modelcontextprotocol/server/transport/StdioServerTransportProviderTests.java
Show comments
View file
Open in desktop
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters. Learn more about bidirectional Unicode characters
Show hidden characters
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import java.util.Map;
import java.util.concurrent.CountDownLatch;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicReference;

import io.modelcontextprotocol.json.McpJsonDefaults;
Expand All @@ -30,6 +31,7 @@
import org.junit.jupiter.api.Test;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.publisher.Sinks;
import reactor.core.scheduler.Schedulers;
import reactor.test.StepVerifier;

Expand Down Expand Up @@ -103,6 +105,48 @@ void shouldCreateSessionWhenSessionFactoryIsSet() {
assertThat(testErr.toString()).doesNotContain("Error");
}

/**
* Regression test for https://github.com/modelcontextprotocol/java-sdk/issues/303.
*
* {@code StdioMcpSessionTransport#sendMessage} gates the actual write on
* {@code inboundReady} and {@code outboundReady} (two {@code Sinks.One<Void>}) both
* completing before it proceeds. This mirrors that exact combinator: {@code
* Mono.zip} completes as soon as the FIRST of two value-less {@code Mono<Void>}
* sources completes -- it never actually waits for the second one -- silently
* breaking the readiness barrier. {@code Mono.when} is the correct operator for
* waiting on multiple completion-only signals and does wait for both.
*/
@Test
void monoZipDoesNotWaitForBothVoidSignals_monoWhenDoes() {
Sinks.One<Void> inboundReady = Sinks.one();
Sinks.One<Void> outboundReady = Sinks.one();
AtomicBoolean zipProceeded = new AtomicBoolean(false);

Mono.zip(inboundReady.asMono(), outboundReady.asMono())
.then(Mono.fromRunnable(() -> zipProceeded.set(true)))
.subscribe();

// Only outboundReady has fired; inboundReady is still pending.
outboundReady.tryEmitValue(null);

assertThat(zipProceeded).as("Mono.zip incorrectly proceeds after only ONE of the two signals fires").isTrue();

Sinks.One<Void> inboundReady2 = Sinks.one();
Sinks.One<Void> outboundReady2 = Sinks.one();
AtomicBoolean whenProceeded = new AtomicBoolean(false);

Mono.when(inboundReady2.asMono(), outboundReady2.asMono())
.then(Mono.fromRunnable(() -> whenProceeded.set(true)))
.subscribe();

// Only outboundReady2 has fired; inboundReady2 is still pending.
outboundReady2.tryEmitValue(null);
assertThat(whenProceeded).as("Mono.when must not proceed until BOTH signals are ready").isFalse();

inboundReady2.tryEmitValue(null);
assertThat(whenProceeded).as("Mono.when must proceed once BOTH signals are ready").isTrue();
}

@Test
void shouldHandleIncomingMessages() {

Expand Down
Back | FazBrowse Home | New Git URL