From 1d5d6deef77805b4c63662b98f6818818573bb8e Mon Sep 17 00:00:00 2001 From: kamisheng <1659225248@qq.com> Date: Mon, 12 Oct 2026 21:18:59 +0800 Subject: [PATCH] Create the reduce accumulator per subscription in the list* methods The Mono returned by listTools, listResources, listResourceTemplates, listPrompts and listRoots captured a single mutable accumulator created when the method was called, so every subscription kept appending to it: Mono listRoots = exchange.listRoots(); listRoots.block().roots().size(); // 2 listRoots.block().roots().size(); // 4 listRoots.block().roots().size(); // 6 Concurrent subscriptions also raced on the shared ArrayList. Use reduceWith(Supplier, accumulator) so that a fresh accumulator is created for every subscription. A single subscription behaves exactly as before. --- .../client/McpAsyncClient.java | 8 +- .../server/McpAsyncServerExchange.java | 2 +- .../server/McpAsyncServerExchangeTests.java | 21 +++++ .../client/AbstractMcpAsyncClientTests.java | 80 +++++++++++++++++++ 4 files changed, 106 insertions(+), 5 deletions(-) diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java b/mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java index 3509b760b..502b23f6b 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/client/McpAsyncClient.java @@ -734,7 +734,7 @@ public Mono listTools() { return this.listTools(McpSchema.FIRST_PAGE).expand(result -> { String next = result.nextCursor(); return (next != null && !next.isEmpty()) ? this.listTools(next) : Mono.empty(); - }).reduce(new ArrayList(), (accumulated, result) -> { + }).reduceWith(ArrayList::new, (accumulated, result) -> { accumulated.addAll(result.tools()); return accumulated; }).map(all -> McpSchema.ListToolsResult.builder(Collections.unmodifiableList(all)).build()); @@ -821,7 +821,7 @@ public Mono listResources() { return this.listResources(McpSchema.FIRST_PAGE).expand(result -> { String next = result.nextCursor(); return (next != null && !next.isEmpty()) ? this.listResources(next) : Mono.empty(); - }).reduce(new ArrayList(), (accumulated, result) -> { + }).reduceWith(ArrayList::new, (accumulated, result) -> { accumulated.addAll(result.resources()); return accumulated; }).map(all -> McpSchema.ListResourcesResult.builder(Collections.unmodifiableList(all)).build()); @@ -907,7 +907,7 @@ public Mono listResourceTemplates() { return this.listResourceTemplates(McpSchema.FIRST_PAGE).expand(result -> { String next = result.nextCursor(); return (next != null && !next.isEmpty()) ? this.listResourceTemplates(next) : Mono.empty(); - }).reduce(new ArrayList(), (accumulated, result) -> { + }).reduceWith(ArrayList::new, (accumulated, result) -> { accumulated.addAll(result.resourceTemplates()); return accumulated; }).map(all -> McpSchema.ListResourceTemplatesResult.builder(Collections.unmodifiableList(all)).build()); @@ -1026,7 +1026,7 @@ public Mono listPrompts() { return this.listPrompts(McpSchema.FIRST_PAGE).expand(result -> { String next = result.nextCursor(); return (next != null && !next.isEmpty()) ? this.listPrompts(next) : Mono.empty(); - }).reduce(new ArrayList(), (accumulated, result) -> { + }).reduceWith(ArrayList::new, (accumulated, result) -> { accumulated.addAll(result.prompts()); return accumulated; }).map(all -> McpSchema.ListPromptsResult.builder(Collections.unmodifiableList(all)).build()); diff --git a/mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java b/mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java index e27d6128f..280a3c87a 100644 --- a/mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java +++ b/mcp-core/src/main/java/io/modelcontextprotocol/server/McpAsyncServerExchange.java @@ -209,7 +209,7 @@ public Mono listRoots() { return this.listRoots(McpSchema.FIRST_PAGE) .expand(result -> (result.nextCursor() != null) ? this.listRoots(result.nextCursor()) : Mono.empty()) - .reduce(McpSchema.ListRootsResult.builder(new ArrayList<>()).build(), + .reduceWith(() -> McpSchema.ListRootsResult.builder(new ArrayList<>()).build(), (allRootsResult, result) -> { allRootsResult.roots().addAll(result.roots()); return allRootsResult; diff --git a/mcp-core/src/test/java/io/modelcontextprotocol/server/McpAsyncServerExchangeTests.java b/mcp-core/src/test/java/io/modelcontextprotocol/server/McpAsyncServerExchangeTests.java index f4f76b159..d56829a74 100644 --- a/mcp-core/src/test/java/io/modelcontextprotocol/server/McpAsyncServerExchangeTests.java +++ b/mcp-core/src/test/java/io/modelcontextprotocol/server/McpAsyncServerExchangeTests.java @@ -85,6 +85,27 @@ void testListRootsWithSinglePage() { }).verifyComplete(); } + @Test + void testListRootsCanBeResubscribed() { + + List roots = Arrays + .asList(McpSchema.Root.builder("file:///home/user/project1").name("Project 1").build()); + McpSchema.ListRootsResult singlePageResult = McpSchema.ListRootsResult.builder(roots).build(); + + when(mockSession.sendRequest(eq(McpSchema.METHOD_ROOTS_LIST), any(McpSchema.PaginatedRequest.class), + any(TypeRef.class))) + .thenReturn(Mono.just(singlePageResult)); + + Mono listRoots = exchange.listRoots(); + + // The same Mono must be safe to subscribe more than once: the accumulator used to + // aggregate the pages has to be created per subscription, not once when + // listRoots() + // is called. + StepVerifier.create(listRoots).assertNext(result -> assertThat(result.roots()).hasSize(1)).verifyComplete(); + StepVerifier.create(listRoots).assertNext(result -> assertThat(result.roots()).hasSize(1)).verifyComplete(); + } + @Test void testListRootsWithMultiplePages() { diff --git a/mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java b/mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java index f422871b0..55a98c430 100644 --- a/mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java +++ b/mcp-test/src/main/java/io/modelcontextprotocol/client/AbstractMcpAsyncClientTests.java @@ -206,6 +206,26 @@ void testListAllTools() { }); } + @Test + void testListAllToolsCanBeResubscribed() { + withClient(createMcpTransport(), mcpAsyncClient -> { + Mono listTools = mcpAsyncClient.listTools(); + + AtomicInteger firstSubscriptionSize = new AtomicInteger(); + StepVerifier.create(mcpAsyncClient.initialize().then(listTools)).consumeNextWith(result -> { + firstSubscriptionSize.set(result.tools().size()); + assertThat(result.tools()).isNotEmpty(); + }).verifyComplete(); + + // The same Mono must be safe to subscribe again: every subscription has to + // start from an empty accumulator instead of appending to the previous + // result. + StepVerifier.create(listTools) + .consumeNextWith(result -> assertThat(result.tools()).hasSize(firstSubscriptionSize.get())) + .verifyComplete(); + }); + } + @Test void testListAllToolsReturnsImmutableList() { withClient(createMcpTransport(), mcpAsyncClient -> { @@ -378,6 +398,26 @@ void testListAllResources() { }); } + @Test + void testListAllResourcesCanBeResubscribed() { + withClient(createMcpTransport(), mcpAsyncClient -> { + Mono listResources = mcpAsyncClient.listResources(); + + AtomicInteger firstSubscriptionSize = new AtomicInteger(); + StepVerifier.create(mcpAsyncClient.initialize().then(listResources)).consumeNextWith(result -> { + firstSubscriptionSize.set(result.resources().size()); + assertThat(result.resources()).isNotEmpty(); + }).verifyComplete(); + + // The same Mono must be safe to subscribe again: every subscription has to + // start from an empty accumulator instead of appending to the previous + // result. + StepVerifier.create(listResources) + .consumeNextWith(result -> assertThat(result.resources()).hasSize(firstSubscriptionSize.get())) + .verifyComplete(); + }); + } + @Test void testListAllResourcesReturnsImmutableList() { withClient(createMcpTransport(), mcpAsyncClient -> { @@ -443,6 +483,26 @@ void testListAllPrompts() { }); } + @Test + void testListAllPromptsCanBeResubscribed() { + withClient(createMcpTransport(), mcpAsyncClient -> { + Mono listPrompts = mcpAsyncClient.listPrompts(); + + AtomicInteger firstSubscriptionSize = new AtomicInteger(); + StepVerifier.create(mcpAsyncClient.initialize().then(listPrompts)).consumeNextWith(result -> { + firstSubscriptionSize.set(result.prompts().size()); + assertThat(result.prompts()).isNotEmpty(); + }).verifyComplete(); + + // The same Mono must be safe to subscribe again: every subscription has to + // start from an empty accumulator instead of appending to the previous + // result. + StepVerifier.create(listPrompts) + .consumeNextWith(result -> assertThat(result.prompts()).hasSize(firstSubscriptionSize.get())) + .verifyComplete(); + }); + } + @Test void testListAllPromptsReturnsImmutableList() { withClient(createMcpTransport(), mcpAsyncClient -> { @@ -636,6 +696,26 @@ void testListAllResourceTemplates() { }); } + @Test + void testListAllResourceTemplatesCanBeResubscribed() { + withClient(createMcpTransport(), mcpAsyncClient -> { + Mono listResourceTemplates = mcpAsyncClient.listResourceTemplates(); + + AtomicInteger firstSubscriptionSize = new AtomicInteger(); + StepVerifier.create(mcpAsyncClient.initialize().then(listResourceTemplates)).consumeNextWith(result -> { + firstSubscriptionSize.set(result.resourceTemplates().size()); + assertThat(result.resourceTemplates()).isNotEmpty(); + }).verifyComplete(); + + // The same Mono must be safe to subscribe again: every subscription has to + // start from an empty accumulator instead of appending to the previous + // result. + StepVerifier.create(listResourceTemplates) + .consumeNextWith(result -> assertThat(result.resourceTemplates()).hasSize(firstSubscriptionSize.get())) + .verifyComplete(); + }); + } + @Test void testListAllResourceTemplatesReturnsImmutableList() { withClient(createMcpTransport(), mcpAsyncClient -> {