StdioServerTransportProvider uses incorrect Mono.zip operator

[damien-urruty-sonarsource]

Hello,

I was debugging another issue and looking at the code, and I noticed this line: [#[line]](

java-sdk/mcp/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java

Line 152 in 2f94434

return Mono.zip(inboundReady.asMono(), outboundReady.asMono()).then(Mono.defer(() -> {

)

When opened in IntelliJ, I noticed this warning being reported:

I am not experienced at all with Mono so I don't know if it's a real problem or not, sharing just in case...

[chemicL]

Thanks for the heads up. I think the warnings come from Intellij Idea and I think they are inaccurate. Perhaps you'd like to report it in their issue tracker?

However, I think the used operator is probably not the right one and when should be used instead. Compare:

public static void main(String[] args) throws InterruptedException {
		Mono.zip(
				Mono.delay(Duration.ofSeconds(2)).then().doOnCancel(() -> System.out.println("Slower mono cancelled!")),
				Mono.delay(Duration.ofSeconds(1)).then().doOnCancel(() -> System.out.println("Faster mono cancelled!"))
		)
		.subscribe(
				x -> System.out.println("Value: " + x),
				e -> System.err.println("Error: "+ e),
				() -> System.out.println("Completed")
		);

		Thread.sleep(3000);
	}
}

// prints after 1s:
// Slower mono cancelled!
// Completed

with

public static void main(String[] args) throws InterruptedException {
		Mono.when(
				Mono.delay(Duration.ofSeconds(2)).then().doOnCancel(() -> System.out.println("Slower mono cancelled!")),
				Mono.delay(Duration.ofSeconds(1)).then().doOnCancel(() -> System.out.println("Faster mono cancelled!"))
		)
		.subscribe(
				x -> System.out.println("Value: " + x),
				e -> System.err.println("Error: "+ e),
				() -> System.out.println("Completed")
		);

		Thread.sleep(3000);
	}
}

// prints after 2s:
// Completed

So for this particular use case I believe we should wait for both inbound and outbound streams. Let me mark the issue as a good first issue.

[imankhadiv]

Hi @chemicL — is this still open for a fresh PR?

The issue is still present on main: StdioServerTransportProvider uses Mono.zip(inboundReady.asMono(), outboundReady.asMono()), and both inboundReady/outboundReady are Sinks.One<Void> that only ever complete empty. Mono.zip requires a value from every source, so the first empty completion completes the combined Mono and cancels the other one — meaning sendMessage can proceed before both streams are actually ready. Mono.when waits for both without cancelling, as you pointed out above.

There is an existing PR (#846) with this change, but it has had no review since it was opened in March, and two other contributors have since opened and self-closed duplicates deferring to it (#981, #987). If #846 is still the intended path, I will leave this alone. If it is stalled, I would be happy to open a fresh PR with the Mono.when fix plus a regression test that fails on zip and passes on when.