修复Get结束Emitter不close的问题

Author: relat-ivity

framework/fel/java/plugins/tool-mcp-server/src/main/java/modelengine/fel/tool/mcp/server/DefaultMcpServer.java

@@ -6,22 +6,19 @@
 
 package modelengine.fel.tool.mcp.server;
 
-import com.fasterxml.jackson.databind.ObjectMapper;
 import io.modelcontextprotocol.server.McpServerFeatures;
 import io.modelcontextprotocol.server.McpSyncServerExchange;
 import io.modelcontextprotocol.spec.McpSchema;
 import modelengine.fel.tool.mcp.entity.ServerSchema;
 import modelengine.fel.tool.mcp.entity.Tool;
 import modelengine.fel.tool.service.ToolChangedObserver;
 import modelengine.fel.tool.service.ToolExecuteService;
-import modelengine.fitframework.annotation.Bean;
 import modelengine.fitframework.annotation.Component;
 import io.modelcontextprotocol.server.McpSyncServer;
 import modelengine.fitframework.log.Logger;
 import modelengine.fitframework.util.MapUtils;
 import modelengine.fitframework.util.StringUtils;
 
-import java.time.Duration;
 import java.util.ArrayList;
 import java.util.List;
 import java.util.Map;
@@ -34,7 +31,8 @@
 import static modelengine.fitframework.inspection.Validation.notNull;
 
 /**
- * Mcp Server implementing interface {@link McpServer}, {@link ToolChangedObserver} with MCP SDK.
+ * Mcp Server implementing interface {@link McpServer}, {@link ToolChangedObserver}
+ * with MCP Server Bean {@link McpSyncServer}.
  *
  * @author 黄可欣
  * @since 2025-09-30

framework/fel/java/plugins/tool-mcp-server/src/main/java/modelengine/fel/tool/mcp/server/DefaultMcpStreamableServerTransportProvider.java

@@ -278,14 +278,17 @@ public void onEmittedData(TextEvent data) {
                         @Override
                         public void onCompleted() {
                             logger.info("[SSE] Completed SSE emitting for session: {}", sessionId);
-                            listeningStream.close();
                         }
 
                         @Override
                         public void onFailed(Exception cause) {
-                            // No action needed
+                            logger.warn("[SSE] SSE failed for session: {}, cause: {}", sessionId, cause.getMessage());
                         }
                     });
+                    
+                    // Add connection monitoring to detect client disconnection
+                    // This is a workaround to ensure listeningStream.close() is called when client disconnects
+                    startConnectionMonitoring(sessionId, listeningStream, response);
                 }
             });
         }
@@ -296,6 +299,43 @@ public void onFailed(Exception cause) {
         }
     }
 
+    /**
+     * Starts connection monitoring to detect client disconnection and ensure proper cleanup.
+     * This is a workaround to ensure listeningStream.close() is called when client disconnects.
+     *
+     * @param sessionId The session ID
+     * @param listeningStream The listening stream to close when connection is lost
+     * @param response The HTTP response to check for connection status
+     */
+    private void startConnectionMonitoring(String sessionId,
+                                        McpStreamableServerSession.McpStreamableServerSessionStream listeningStream,
+                                        HttpClassicServerResponse response) {
+        // Use a separate thread to periodically check connection status
+        Thread monitoringThread = new Thread(() -> {
+            try {
+                while (!Thread.currentThread().isInterrupted()) {
+                    Thread.sleep(1000); // Check every second
+                    
+                    // Check if the HTTP response is still active
+                    if (!response.isActive()) {
+                        logger.info("[SSE] Connection lost for session, completing emitter to trigger cleanup");
+                        listeningStream.close();
+                        break;
+                    }
+                }
+            } catch (InterruptedException e) {
+                logger.debug("[SSE] Connection monitoring interrupted for session");
+                Thread.currentThread().interrupt();
+            } catch (Exception e) {
+                logger.warn("[SSE] Error in connection monitoring: {}", e.getMessage());
+            }
+        });
+        
+        monitoringThread.setDaemon(true);
+        monitoringThread.setName("sse-connection-monitor-" + sessionId);
+        monitoringThread.start();
+    }
+
     /**
      * Handles POST requests for incoming JSON-RPC messages from clients.
      *

framework/fel/java/plugins/tool-mcp-server/src/main/java/modelengine/fel/tool/mcp/server/McpServer.java

@@ -12,7 +12,6 @@
 import modelengine.fel.tool.mcp.entity.Tool;
 
 import java.util.List;
-import java.util.Map;
 import java.util.function.BiFunction;
 
 /**

framework/fel/java/plugins/tool-mcp-server/src/test/java/modelengine/fel/tool/mcp/server/support/DefaultMcpServerTest.java

@@ -13,7 +13,6 @@
 import static org.mockito.Mockito.verify;
 
 import io.modelcontextprotocol.server.McpSyncServer;
-import io.modelcontextprotocol.spec.McpSchema;
 import modelengine.fel.tool.mcp.entity.ServerSchema;
 import modelengine.fel.tool.mcp.entity.Tool;
 import modelengine.fel.tool.mcp.server.DefaultMcpServer;