From 8104a5d34ed3ec50068bb4ec99b481368ac07702 Mon Sep 17 00:00:00 2001 From: liwei1dao Date: Thu, 26 Jun 2025 19:08:44 +0800 Subject: [PATCH] =?UTF-8?q?=E4=B8=8A=E4=BC=A0mac=E6=96=AD=E5=BC=80?= =?UTF-8?q?=E9=93=BE=E6=8E=A5=20=E7=9A=84=E5=A4=84=E7=90=86=E4=BC=98?= =?UTF-8?q?=E5=8C=96?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- .../chat_api/CustomSseClientTransport.kt | 176 ++++++++++++------ .../yunqiinnovation/chat_api/MCPSubClient.kt | 63 ++++++- 2 files changed, 173 insertions(+), 66 deletions(-) diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt index 12982f0fb..aa469e62a 100644 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt +++ b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt @@ -25,6 +25,7 @@ class CustomSseClientTransport( private val urlString: String?, private val reconnectionTime: Duration? = null, private val requestBuilder: HttpRequestBuilder.() -> Unit = {}, + private val onConnectionLost: (() -> Unit)? = null ) : AbstractTransport() { private val TAG = "CustomSseClientTransport" @@ -35,8 +36,10 @@ class CustomSseClientTransport( private val initialized = AtomicBoolean(false) private var session: ClientSSESession by Delegates.notNull() private val endpoint = CompletableDeferred() + private val isConnected = AtomicBoolean(false) private var job: Job? = null + private var connectionMonitorJob: Job? = null // 创建JSON解析器 private val json = Json { @@ -95,74 +98,118 @@ class CustomSseClientTransport( */ private suspend fun collectEvents() { job = scope.launch(CoroutineName("CustomSseMcpClientTransport.collect#${hashCode()}")) { - session.incoming.collect { event -> - when (event.event) { - "error" -> { - val e = IllegalStateException("SSE error: ${event.data}") - Log.e(TAG, "SSE错误: ${event.data}") - _onError(e) - throw e - } + try { + session.incoming.collect { event -> + when (event.event) { + "error" -> { + Log.e(TAG, "SSE错误: ${event.data}") + isConnected.set(false) + val exception = Exception("SSE Error: ${event.data}") + _onError(exception) + onConnectionLost?.invoke() + throw exception + } - "open" -> { - // SSE连接已打开 - } - "ping" -> { - // 心跳 - } - "endpoint" -> { - try { - val eventData = event.data ?: "" + "open" -> { + // SSE连接已打开 + Log.d(TAG, "SSE连接已打开") + isConnected.set(true) + } + "ping" -> { + // 心跳 + } + "endpoint" -> { + try { + val eventData = event.data ?: "" - // 构建完整的端点URL - val fullEndpoint = if (eventData.contains(hostPart)) { - eventData - } else if (eventData.startsWith("/")) { - "$hostPart$eventData" - } else { - eventData - } - - // 添加查询参数 - val endpointWithParams = if (queryParams.isNotEmpty()) { - if (fullEndpoint.contains("?")) { - val queryString = queryParams.entries.joinToString("&") { "${it.key}=${it.value}" } - "$fullEndpoint&$queryString" + // 构建完整的端点URL + val fullEndpoint = if (eventData.contains(hostPart)) { + eventData + } else if (eventData.startsWith("/")) { + "$hostPart$eventData" } else { - val queryString = queryParams.entries.joinToString("&") { "${it.key}=${it.value}" } - "$fullEndpoint?$queryString" + eventData } - } else { - fullEndpoint + + // 添加查询参数 + val endpointWithParams = if (queryParams.isNotEmpty()) { + if (fullEndpoint.contains("?")) { + val queryString = queryParams.entries.joinToString("&") { "${it.key}=${it.value}" } + "$fullEndpoint&$queryString" + } else { + val queryString = queryParams.entries.joinToString("&") { "${it.key}=${it.value}" } + "$fullEndpoint?$queryString" + } + } else { + fullEndpoint + } + + endpoint.complete(endpointWithParams) + } catch (e: Exception) { + Log.e(TAG, "处理endpoint事件失败: ${e.message}", e) + _onError(e) + close() + error(e) } - - endpoint.complete(endpointWithParams) - } catch (e: Exception) { - Log.e(TAG, "处理endpoint事件失败: ${e.message}", e) - _onError(e) - close() - error(e) } - } - else -> { - try { - val data = event.data - if (data != null) { - try { - val message = json.decodeFromString(data) - _onMessage(message) - } catch (e: Exception) { - Log.e(TAG, "解析JSON-RPC消息失败: ${e.message}", e) - _onError(e) + else -> { + try { + val data = event.data + if (data != null) { + try { + val message = json.decodeFromString(data) + _onMessage(message) + } catch (e: Exception) { + Log.e(TAG, "解析JSON-RPC消息失败: ${e.message}", e) + _onError(e) + } } + } catch (e: Exception) { + Log.e(TAG, "处理事件失败: ${e.message}", e) + _onError(e) } - } catch (e: Exception) { - Log.e(TAG, "处理事件失败: ${e.message}", e) - _onError(e) } } } + } catch (e: CancellationException) { + Log.d(TAG, "SSE事件收集被取消") + throw e + } catch (e: Exception) { + Log.e(TAG, "SSE连接异常断开: ${e.message}", e) + isConnected.set(false) + _onError(e) + onConnectionLost?.invoke() + throw e + } + } + + // 启动连接监控 + startConnectionMonitor() + } + + /** + * 启动连接监控，定期检查连接状态 + */ + private fun startConnectionMonitor() { + connectionMonitorJob = scope.launch { + while (isActive && isConnected.get()) { + try { + delay(10000) // 每10秒检查一次 + + // 检查session是否仍然活跃 + if (session.coroutineContext[Job]?.isCancelled == true) { + Log.w(TAG, "检测到SSE会话已取消") + isConnected.set(false) + onConnectionLost?.invoke() + break + } + } catch (e: Exception) { + Log.e(TAG, "连接监控异常: ${e.message}", e) + isConnected.set(false) + onConnectionLost?.invoke() + break + } } } } @@ -212,6 +259,8 @@ class CustomSseClientTransport( // 等待endpoint就绪 endpoint.await() + + Log.d(TAG, "CustomSseClientTransport启动完成") } /** @@ -245,6 +294,13 @@ class CustomSseClientTransport( } } + /** + * 检查连接状态 + */ + fun isConnectionActive(): Boolean { + return isConnected.get() && session.coroutineContext[Job]?.isActive == true + } + /** * 关闭传输层 */ @@ -253,8 +309,14 @@ class CustomSseClientTransport( Log.e(TAG, "关闭失败: 传输层未初始化") error("CustomSseClientTransport is not initialized!") } + + isConnected.set(false) + connectionMonitorJob?.cancel() session.cancel() _onClose() job?.cancelAndJoin() + connectionMonitorJob?.cancelAndJoin() + + Log.d(TAG, "CustomSseClientTransport已关闭") } -} \ No newline at end of file +} \ No newline at end of file diff --git a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt index 77393a44a..d04ce3bb6 100644 --- a/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt +++ b/local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt @@ -44,13 +44,14 @@ class MCPSubClient( private var mcpClient: Client? = null private var isConnected = false private var availableTools = mutableListOf() + private var transport: CustomSseClientTransport? = null /** * 连接到MCP服务器 */ suspend fun connect(): Boolean = connectionMutex.withLock { if (isConnected) return true - + Log.e(TAG, "[$serverId] 开始连接mcp服务器: $serverUrl") return try { // 创建MCP客户端实例 val client = Client( @@ -61,13 +62,20 @@ class MCPSubClient( ) // 根据URL类型选择传输方式 - val transport = when { + val newTransport = when { serverUrl.startsWith("http://") || serverUrl.startsWith("https://") -> { // SSE传输 - 使用自定义的CustomSseClientTransport val mcpHttpClient = httpClient ?: createMcpHttpClient() CustomSseClientTransport( client = mcpHttpClient, - urlString = serverUrl + urlString = serverUrl, + onConnectionLost = { + // 连接断开回调 + Log.w(TAG, "[$serverId] 检测到连接断开") + scope.launch { + handleConnectionLost() + } + } ) } else -> { @@ -76,8 +84,10 @@ class MCPSubClient( } } + transport = newTransport + // 连接到服务器 - client.connect(transport) + client.connect(newTransport) // 获取可用工具列表 try { @@ -101,9 +111,9 @@ class MCPSubClient( retryCount = 0 currentReconnectDelay = initialReconnectDelay // 启动心跳检测 - startHeartbeat() +// startHeartbeat() + Log.e(TAG, "[$serverId] 连接mcp服务器成功: $serverUrl") true - } catch (e: Exception) { Log.e(TAG, "[$serverId] MCP连接失败: ${e.message}", e) false @@ -321,7 +331,15 @@ class MCPSubClient( */ suspend fun checkConnection(): Boolean { if (!isConnected) { - Log.d(TAG, "当前未连接，尝试重新连接...") + Log.d(TAG, "[$serverId] 当前未连接，尝试重新连接...") + return connect() + } + + // 检查传输层连接状态 + val transportActive = transport?.isConnectionActive() ?: false + if (!transportActive) { + Log.w(TAG, "[$serverId] 传输层连接已断开") + isConnected = false return connect() } @@ -330,11 +348,34 @@ class MCPSubClient( mcpClient?.ping() return true } catch (e: Exception) { - Log.e(TAG, "连接检查失败: ${e.message}") + Log.e(TAG, "[$serverId] 连接检查失败: ${e.message}") isConnected = false return false } } + + /** + * 处理连接断开事件 + */ + private suspend fun handleConnectionLost() { + connectionMutex.withLock { + if (isConnected) { + Log.w(TAG, "[$serverId] 连接已断开，更新状态") + isConnected = false + stopHeartbeat() + + // 可以在这里添加自动重连逻辑 + // 或者通知上层应用连接已断开 + } + } + } + + /** + * 获取连接状态 + */ + fun getConnectionStatus(): Boolean { + return isConnected && (transport?.isConnectionActive() ?: false) + } /** * 停止心跳检测 */ @@ -393,10 +434,14 @@ class MCPSubClient( scope.launch { connectionMutex.withLock { try { + stopHeartbeat() mcpClient?.close() + transport?.close() mcpClient = null + transport = null isConnected = false availableTools.clear() + Log.d(TAG, "[$serverId] MCP连接已关闭") } catch (e: Exception) { Log.e(TAG, "[$serverId] 关闭MCP连接时出错: ${e.message}", e) } @@ -404,4 +449,4 @@ class MCPSubClient( } scope.cancel() } -} \ No newline at end of file +} \ No newline at end of file