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

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

weicu
liwei1dao 1 year ago
parent
54ab644974
commit
ebbd837b7b
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
Split 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 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<String>()
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<JSONRPCMessage>(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<JSONRPCMessage>(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已关闭")
}
}
}

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 isConnected = false
private var availableTools = mutableListOf<Tool>()
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()
}
}
}
Write Preview
Loading…
Cancel
Save