test: consume task streams before assertions

Author: LucaButBoring

mcp-core/src/test/java/io/modelcontextprotocol/server/AbstractMcpClientServerIntegrationTests.java

@@ -21,7 +21,6 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import io.modelcontextprotocol.client.McpClient;
 import io.modelcontextprotocol.client.McpSyncClient;
@@ -1883,8 +1882,7 @@ protected McpSyncClient createTaskClient(String clientType, String name, ClientC
 	}
 
 	/** Extracts the task ID from a list of response messages. */
-	protected String extractTaskId(Stream<ResponseMessage<CallToolResult>> messageStream) {
-		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
+	protected String extractTaskId(List<ResponseMessage<CallToolResult>> messages) {
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
 				return tcm.task().taskId();
@@ -1894,8 +1892,7 @@ protected String extractTaskId(Stream<ResponseMessage<CallToolResult>> messageSt
 	}
 
 	/** Extracts all task statuses from a list of response messages. */
-	protected List<TaskStatus> extractTaskStatuses(Stream<ResponseMessage<CallToolResult>> messageStream) {
-		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
+	protected List<TaskStatus> extractTaskStatuses(List<ResponseMessage<CallToolResult>> messages) {
 		List<TaskStatus> statuses = new ArrayList<>();
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
@@ -2018,7 +2015,7 @@ void testElicitationDuringTaskExecution(String clientType) throws InterruptedExc
 			client.initialize();
 
 			var request = new McpSchema.CallToolRequest("needs-input-tool", Map.of(), DEFAULT_TASK_METADATA, null);
-			var messages = client.callToolStream(request);
+			var messages = client.callToolStream(request).toList();
 			var observedStates = extractTaskStatuses(messages);
 
 			assertThat(taskIdRef.get()).as("Task ID should have been set").isNotNull();
@@ -2157,7 +2154,7 @@ void testCreateTaskHandlerWithTaskMetadata(String clientType) {
 
 			// Call with task metadata - should use createTaskHandler directly
 			var request = new McpSchema.CallToolRequest("create-task-tool", Map.of(), DEFAULT_TASK_METADATA, null);
-			var messages = client.callToolStream(request);
+			var messages = client.callToolStream(request).toList();
 
 			assertThat(createTaskHandlerInvoked.get()).as("createTaskHandler should have been invoked").isTrue();
 			assertThat(messages).as("Should have response messages").isNotEmpty();

mcp-test/src/main/java/io/modelcontextprotocol/AbstractMcpClientServerIntegrationTests.java

@@ -22,7 +22,6 @@
 import java.util.function.BiFunction;
 import java.util.function.Function;
 import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 import io.modelcontextprotocol.client.McpClient;
 import io.modelcontextprotocol.client.McpSyncClient;
@@ -1846,8 +1845,7 @@ protected McpSyncClient createTaskClient(String clientType, String name, ClientC
 	}
 
 	/** Extracts the task ID from a list of response messages. */
-	protected String extractTaskId(Stream<ResponseMessage<CallToolResult>> messageStream) {
-		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
+	protected String extractTaskId(List<ResponseMessage<CallToolResult>> messages) {
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
 				return tcm.task().taskId();
@@ -1857,8 +1855,7 @@ protected String extractTaskId(Stream<ResponseMessage<CallToolResult>> messageSt
 	}
 
 	/** Extracts all task statuses from a list of response messages. */
-	protected List<TaskStatus> extractTaskStatuses(Stream<ResponseMessage<CallToolResult>> messageStream) {
-		List<ResponseMessage<CallToolResult>> messages = messageStream.toList();
+	protected List<TaskStatus> extractTaskStatuses(List<ResponseMessage<CallToolResult>> messages) {
 		List<TaskStatus> statuses = new ArrayList<>();
 		for (var msg : messages) {
 			if (msg instanceof TaskCreatedMessage<CallToolResult> tcm) {
@@ -2106,7 +2103,7 @@ void testElicitationDuringTaskExecution(String clientType) throws InterruptedExc
 			client.initialize();
 
 			var request = new McpSchema.CallToolRequest("needs-input-tool", Map.of(), DEFAULT_TASK_METADATA, null);
-			var messages = client.callToolStream(request);
+			var messages = client.callToolStream(request).toList();
 			var observedStates = extractTaskStatuses(messages);
 
 			if (taskIdRef.get() != null) {
@@ -2245,7 +2242,7 @@ void testCreateTaskHandlerWithTaskMetadata(String clientType) {
 
 			// Call with task metadata - should use createTaskHandler directly
 			var request = new McpSchema.CallToolRequest("create-task-tool", Map.of(), DEFAULT_TASK_METADATA, null);
-			var messages = client.callToolStream(request);
+			var messages = client.callToolStream(request).toList();
 
 			assertThat(createTaskHandlerInvoked.get()).as("createTaskHandler should have been invoked").isTrue();
 			assertThat(messages).as("Should have response messages").isNotEmpty();