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

Introduce timeout for session sweeping

Browse source 
Signed-off-by: Daniel Garnier-Moiroux <git@garnier.wf>
This commit is contained in:
Daniel Garnier-Moiroux 2026-09-17 21:41:12 +02:00 • committed by Daniel Garnier-Moiroux
commit e3e0f5172a
1 changed files with 19 additions and 16 deletions
Download patch file Download diff file Expand all files Collapse all files

35
mcp-core/src/main/java/io/modelcontextprotocol/server/transport/HttpServletStreamableServerTransportProvider.java
View file

@ -150,6 +150,8 @@ public class HttpServletStreamableServerTransportProvider extends HttpServlet
*/ */
private final Disposable sessionSweeper; private final Disposable sessionSweeper;
private static Duration SESSION_SWEEP_TIMEOUT = Duration.ofSeconds(30);
/** /**
* Security validator for validating HTTP requests. * Security validator for validating HTTP requests.
*/ */
@ -217,14 +219,13 @@ public class HttpServletStreamableServerTransportProvider extends HttpServlet
* down * down
*/ */
private Disposable startSessionSweeper(Duration sessionSweepInterval) { private Disposable startSessionSweeper(Duration sessionSweepInterval) {
return Flux.interval(sessionSweepInterval, sessionSweepInterval, Schedulers.boundedElastic()) return Flux
.interval(sessionSweepInterval, sessionSweepInterval,
Schedulers.newSingle("streamable-http-server-transport-session-sweeper"))
.concatMap( .concatMap(
tick -> sweepSessions().doOnError(e -> logger.error("Session sweep failed", e)).onErrorComplete()) tick -> sweepSessions().doOnError(e -> logger.error("Session sweep failed", e)).onErrorComplete())
.onErrorComplete(error -> { .subscribe(next -> {
logger.error("Session sweeper error", error); }, error -> logger.error("Session sweeper error", error));
return true;
})
.subscribe();
} }
/** /**
@ -240,16 +241,18 @@ public class HttpServletStreamableServerTransportProvider extends HttpServlet
return Mono.empty(); return Mono.empty();
} }
Set<String> active = this.activeSessions.getAndSet(ConcurrentHashMap.newKeySet()); Set<String> active = this.activeSessions.getAndSet(ConcurrentHashMap.newKeySet());
return Flux.fromIterable(this.sessions.values()) return Flux.fromIterable(this.sessions.values()).filter(session -> {
.filter(session -> !session.hasOpenStream() && !active.contains(session.getId())) if (session.hasOpenStream() || active.contains(session.getId())) {
.filter(session -> this.sessions.remove(session.getId(), session)) return false;
.flatMap(session -> { }
logger.debug("Evicting idle session {}", session.getId()); return this.sessions.remove(session.getId(), session);
return session.closeGracefully() }).flatMap(session -> {
.doOnError(e -> logger.warn("Failed to close idle session {}: {}", session.getId(), e.getMessage())) logger.debug("Evicting idle session {}", session.getId());
.onErrorComplete(); return session.closeGracefully()
}) .timeout(SESSION_SWEEP_TIMEOUT)
.then(); .doOnError(e -> logger.warn("Failed to close idle session {}: {}", session.getId(), e.getMessage()))
.onErrorComplete();
}).then();
} }
/** /**