modelcontextprotocol/java-sdk
Watch
1
0
Fork
You've already forked java-sdk
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(server): Fi StdioServerTransportProvider initialization flow (#74)

Browse source 
Extract message processing initialization from StdioMcpSessionTransport constructor into a
separate initProcessing() method.

Signed-off-by: Christian Tzolov <christian.tzolov@broadcom.com>
This commit is contained in:
Christian Tzolov 2025-03-26 12:09:46 +01:00 • committed by Christian Tzolov
commit 96e6d9cc15
1 changed files with 9 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

14
mcp/src/main/java/io/modelcontextprotocol/server/transport/StdioServerTransportProvider.java
View file

@ -93,7 +93,9 @@ public class StdioServerTransportProvider implements McpServerTransportProvider
@Override @Override
public void setSessionFactory(McpServerSession.Factory sessionFactory) { public void setSessionFactory(McpServerSession.Factory sessionFactory) {
// Create a single session for the stdio connection // Create a single session for the stdio connection
this.session = sessionFactory.create(new StdioMcpSessionTransport()); var transport = new StdioMcpSessionTransport();
this.session = sessionFactory.create(transport);
transport.initProcessing();
} }
@Override @Override
@ -142,10 +144,6 @@ public class StdioServerTransportProvider implements McpServerTransportProvider
"stdio-inbound"); "stdio-inbound");
this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(), this.outboundScheduler = Schedulers.fromExecutorService(Executors.newSingleThreadExecutor(),
"stdio-outbound"); "stdio-outbound");
handleIncomingMessages();
startInboundProcessing();
startOutboundProcessing();
} }
@Override @Override
@ -181,6 +179,12 @@ public class StdioServerTransportProvider implements McpServerTransportProvider
logger.debug("Session transport closed"); logger.debug("Session transport closed");
} }
private void initProcessing() {
handleIncomingMessages();
startInboundProcessing();
startOutboundProcessing();
}
private void handleIncomingMessages() { private void handleIncomingMessages() {
this.inboundSink.asFlux().flatMap(message -> session.handle(message)).doOnTerminate(() -> { this.inboundSink.asFlux().flatMap(message -> session.handle(message)).doOnTerminate(() -> {
// The outbound processing will dispose its scheduler upon completion // The outbound processing will dispose its scheduler upon completion