liwei1dao
/
deepvoice_client
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity
Browse Source

上传mac断开链接 的处理优化

newdev_shunjiawei
liwei1dao 1 year ago
parent
e223e436b4
commit
8104a5d34e
2 changed files with 173 additions and 66 deletions
Whitespace
Show all changes Ignore whitespace when comparing lines Ignore changes in amount of whitespace Ignore changes in whitespace at EOL
Unified View
Diff Options
Show Stats Download Patch File Download Diff File
  1. 176
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt
  2. 63
      local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt

176
local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/CustomSseClientTransport.kt
View File

@ -25,6 +25,7 @@ class CustomSseClientTransport(
private val urlString: String?, private val urlString: String?,
private val reconnectionTime: Duration? = null, private val reconnectionTime: Duration? = null,
private val requestBuilder: HttpRequestBuilder.() -> Unit = {}, private val requestBuilder: HttpRequestBuilder.() -> Unit = {},
private val onConnectionLost: (() -> Unit)? = null
) : AbstractTransport() { ) : AbstractTransport() {
private val TAG = "CustomSseClientTransport" private val TAG = "CustomSseClientTransport"
@ -35,8 +36,10 @@ class CustomSseClientTransport(
private val initialized = AtomicBoolean(false) private val initialized = AtomicBoolean(false)
private var session: ClientSSESession by Delegates.notNull() private var session: ClientSSESession by Delegates.notNull()
private val endpoint = CompletableDeferred<String>() private val endpoint = CompletableDeferred<String>()
private val isConnected = AtomicBoolean(false)
private var job: Job? = null private var job: Job? = null
private var connectionMonitorJob: Job? = null
// 创建JSON解析器 // 创建JSON解析器
private val json = Json { private val json = Json {
@ -95,74 +98,118 @@ class CustomSseClientTransport(
*/ */
private suspend fun collectEvents() { private suspend fun collectEvents() {
job = scope.launch(CoroutineName("CustomSseMcpClientTransport.collect#${hashCode()}")) { job = scope.launch(CoroutineName("CustomSseMcpClientTransport.collect#${hashCode()}")) {
session.incoming.collect { event -> try {
when (event.event) { session.incoming.collect { event ->
"error" -> { when (event.event) {
val e = IllegalStateException("SSE error: ${event.data}") "error" -> {
Log.e(TAG, "SSE错误: ${event.data}") Log.e(TAG, "SSE错误: ${event.data}")
_onError(e) isConnected.set(false)
throw e val exception = Exception("SSE Error: ${event.data}")
} _onError(exception)
onConnectionLost?.invoke()
throw exception
}
"open" -> { "open" -> {
// SSE连接已打开 // SSE连接已打开
} Log.d(TAG, "SSE连接已打开")
"ping" -> { isConnected.set(true)
// 心跳 }
} "ping" -> {
"endpoint" -> { // 心跳
try { }
val eventData = event.data ?: "" "endpoint" -> {
try {
val eventData = event.data ?: ""
// 构建完整的端点URL // 构建完整的端点URL
val fullEndpoint = if (eventData.contains(hostPart)) { val fullEndpoint = if (eventData.contains(hostPart)) {
eventData eventData
} else if (eventData.startsWith("/")) { } else if (eventData.startsWith("/")) {
"$hostPart$eventData" "$hostPart$eventData"
} else {
eventData
}
// 添加查询参数
val endpointWithParams = if (queryParams.isNotEmpty()) {
if (fullEndpoint.contains("?")) {
val queryString = queryParams.entries.joinToString("&") { "${it.key}=${it.value}" }
"$fullEndpoint&$queryString"
} else { } else {
val queryString = queryParams.entries.joinToString("&") { "${it.key}=${it.value}" } eventData
"$fullEndpoint?$queryString"
} }
} 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 -> { else -> {
try { try {
val data = event.data val data = event.data
if (data != null) { if (data != null) {
try { try {
val message = json.decodeFromString<JSONRPCMessage>(data) val message = json.decodeFromString<JSONRPCMessage>(data)
_onMessage(message) _onMessage(message)
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "解析JSON-RPC消息失败: ${e.message}", e) Log.e(TAG, "解析JSON-RPC消息失败: ${e.message}", e)
_onError(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就绪
endpoint.await() 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, "关闭失败: 传输层未初始化") Log.e(TAG, "关闭失败: 传输层未初始化")
error("CustomSseClientTransport is not initialized!") error("CustomSseClientTransport is not initialized!")
} }
isConnected.set(false)
connectionMonitorJob?.cancel()
session.cancel() session.cancel()
_onClose() _onClose()
job?.cancelAndJoin() job?.cancelAndJoin()
connectionMonitorJob?.cancelAndJoin()
Log.d(TAG, "CustomSseClientTransport已关闭")
} }
} }

63
local_plugins/chat_api/android/src/main/kotlin/com/yunqiinnovation/chat_api/MCPSubClient.kt
View File

@ -44,13 +44,14 @@ class MCPSubClient(
private var mcpClient: Client? = null private var mcpClient: Client? = null
private var isConnected = false private var isConnected = false
private var availableTools = mutableListOf<Tool>() private var availableTools = mutableListOf<Tool>()
private var transport: CustomSseClientTransport? = null
/** /**
* 连接到MCP服务器 * 连接到MCP服务器
*/ */
suspend fun connect(): Boolean = connectionMutex.withLock { suspend fun connect(): Boolean = connectionMutex.withLock {
if (isConnected) return true if (isConnected) return true
Log.e(TAG, "[$serverId] 开始连接mcp服务器: $serverUrl")
return try { return try {
// 创建MCP客户端实例 // 创建MCP客户端实例
val client = Client( val client = Client(
@ -61,13 +62,20 @@ class MCPSubClient(
) )
// 根据URL类型选择传输方式 // 根据URL类型选择传输方式
val transport = when { val newTransport = when {
serverUrl.startsWith("http://") || serverUrl.startsWith("https://") -> { serverUrl.startsWith("http://") || serverUrl.startsWith("https://") -> {
// SSE传输 - 使用自定义的CustomSseClientTransport // SSE传输 - 使用自定义的CustomSseClientTransport
val mcpHttpClient = httpClient ?: createMcpHttpClient() val mcpHttpClient = httpClient ?: createMcpHttpClient()
CustomSseClientTransport( CustomSseClientTransport(
client = mcpHttpClient, client = mcpHttpClient,
urlString = serverUrl urlString = serverUrl,
onConnectionLost = {
// 连接断开回调
Log.w(TAG, "[$serverId] 检测到连接断开")
scope.launch {
handleConnectionLost()
}
}
) )
} }
else -> { else -> {
@ -76,8 +84,10 @@ class MCPSubClient(
} }
} }
transport = newTransport
// 连接到服务器 // 连接到服务器
client.connect(transport) client.connect(newTransport)
// 获取可用工具列表 // 获取可用工具列表
try { try {
@ -101,9 +111,9 @@ class MCPSubClient(
retryCount = 0 retryCount = 0
currentReconnectDelay = initialReconnectDelay currentReconnectDelay = initialReconnectDelay
// 启动心跳检测 // 启动心跳检测
startHeartbeat() // startHeartbeat()
Log.e(TAG, "[$serverId] 连接mcp服务器成功: $serverUrl")
true true
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "[$serverId] MCP连接失败: ${e.message}", e) Log.e(TAG, "[$serverId] MCP连接失败: ${e.message}", e)
false false
@ -321,7 +331,15 @@ class MCPSubClient(
*/ */
suspend fun checkConnection(): Boolean { suspend fun checkConnection(): Boolean {
if (!isConnected) { 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() return connect()
} }
@ -330,11 +348,34 @@ class MCPSubClient(
mcpClient?.ping() mcpClient?.ping()
return true return true
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "连接检查失败: ${e.message}") Log.e(TAG, "[$serverId] 连接检查失败: ${e.message}")
isConnected = false isConnected = false
return 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 { scope.launch {
connectionMutex.withLock { connectionMutex.withLock {
try { try {
stopHeartbeat()
mcpClient?.close() mcpClient?.close()
transport?.close()
mcpClient = null mcpClient = null
transport = null
isConnected = false isConnected = false
availableTools.clear() availableTools.clear()
Log.d(TAG, "[$serverId] MCP连接已关闭")
} catch (e: Exception) { } catch (e: Exception) {
Log.e(TAG, "[$serverId] 关闭MCP连接时出错: ${e.message}", e) Log.e(TAG, "[$serverId] 关闭MCP连接时出错: ${e.message}", e)
} }
@ -404,4 +449,4 @@ class MCPSubClient(
} }
scope.cancel() scope.cancel()
} }
} }
Write Preview
Loading…
Cancel
Save