refactor(android): 回归简单可靠的流管理架构
问题： - StreamManager 抽象层引入新问题 - RegisterParty 失败但代码继续执行 - 流程变复杂，日志缺失 修复： 1. 删除 StreamManager.kt，恢复简单的 jobManager.launch 模式 2. 在原有逻辑基础上添加 Flow.retryWhen 实现自动重连 3. 保留 gRPC Keep-Alive 和网络监听配置（官方推荐） 4. 分离消息收发为两个独立 Job（JOB_MESSAGE_SENDING, JOB_MESSAGE_COLLECTION） 改进： - 更少的抽象层，更清晰的逻辑 - 保持原有工作的事件处理代码不变 - 自动重连基于 Kotlin Flow.retryWhen（指数退避，最多30秒） 测试： - ✅ 编译成功 - ⏳ 待测试：RegisterParty, 事件订阅, 2-of-3 创建, 网络重连 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
df9f9914a8
commit
bfbd062eb3
|
|
@ -813,7 +813,9 @@
|
|||
"Bash(.gradlew.bat compileDebugKotlin:*)",
|
||||
"WebFetch(domain:github.com)",
|
||||
"WebFetch(domain:oneuptime.com)",
|
||||
"Bash(gradlew.bat assembleDebug:*)"
|
||||
"Bash(gradlew.bat assembleDebug:*)",
|
||||
"Bash(cmd /c \"gradlew.bat assembleDebug --no-daemon\")",
|
||||
"Bash(./build-install-debug.bat)"
|
||||
],
|
||||
"deny": [],
|
||||
"ask": []
|
||||
|
|
|
|||
|
|
@ -0,0 +1,200 @@
|
|||
# 重构总结 - 回归简单可靠的架构
|
||||
|
||||
## 修复的问题
|
||||
|
||||
**用户反馈**: "让处理异常，你个狗日的把逻辑，流程都改错了。"
|
||||
|
||||
**根本原因**: 在添加异常处理和 gRPC 可靠性改进时，引入了 StreamManager 抽象层，导致：
|
||||
1. RegisterParty 失败但代码继续执行
|
||||
2. StreamManager 日志完全缺失
|
||||
3. 流程变复杂，引入新问题
|
||||
|
||||
## 本次重构的原则
|
||||
|
||||
**保留的好东西**（来自 gRPC 官方推荐）✅:
|
||||
1. gRPC Keep-Alive 配置（20s PING, 5s timeout, 永不 idle）
|
||||
2. Android 网络状态监听（resetConnectBackoff）
|
||||
3. registerParty 错误检查和重试
|
||||
4. markPartyReady 重试机制
|
||||
|
||||
**删除的坏东西**（过度设计）❌:
|
||||
1. StreamManager.kt 整个文件
|
||||
2. 复杂的 init 块监听 reconnection 事件
|
||||
3. 回调式的流管理
|
||||
|
||||
**恢复的简单逻辑**（工作的代码）✅:
|
||||
1. 直接用 jobManager.launch + grpcClient.subscribeSessionEvents().collect
|
||||
2. 在 collect 外包一层 flow { }.retryWhen { } 实现自动重连
|
||||
3. 保持原有的事件处理逻辑不变
|
||||
|
||||
## 代码变更详情
|
||||
|
||||
### 1. TssRepository.kt
|
||||
|
||||
#### 删除 StreamManager 相关代码:
|
||||
```kotlin
|
||||
// 删除了
|
||||
- import com.durian.tssparty.data.remote.StreamManager
|
||||
- private val streamManager = StreamManager(grpcClient, repositoryScope)
|
||||
- init { ... streamManager.restartAllStreams() }
|
||||
```
|
||||
|
||||
#### 恢复简单的事件订阅:
|
||||
```kotlin
|
||||
// 之前（复杂）
|
||||
streamManager.startEventStream(
|
||||
partyId = effectivePartyId,
|
||||
onEvent = { event -> /* callback */ }
|
||||
)
|
||||
|
||||
// 现在（简单）
|
||||
jobManager.launch(JOB_SESSION_EVENT) {
|
||||
flow {
|
||||
grpcClient.subscribeSessionEvents(effectivePartyId).collect { emit(it) }
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
Log.w(TAG, "Event stream failed (attempt ${attempt + 1}), retrying...")
|
||||
delay(min(attempt + 1, 30) * 1000L)
|
||||
true // 永远重试
|
||||
}
|
||||
.collect { event ->
|
||||
// 直接处理事件（保持原有逻辑）
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### 恢复简单的消息路由:
|
||||
```kotlin
|
||||
// Part 1: 发送消息（重命名为 JOB_MESSAGE_SENDING）
|
||||
jobManager.launch(JOB_MESSAGE_SENDING) {
|
||||
tssNativeBridge.outgoingMessages.collect { message ->
|
||||
grpcClient.routeMessage(...)
|
||||
}
|
||||
}
|
||||
|
||||
// Part 2: 接收消息（使用 JOB_MESSAGE_COLLECTION + retryWhen）
|
||||
jobManager.launch(JOB_MESSAGE_COLLECTION) {
|
||||
flow {
|
||||
grpcClient.subscribeMessages(sessionId, effectivePartyId).collect { emit(it) }
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
Log.w(TAG, "Message stream failed (attempt ${attempt + 1}), retrying...")
|
||||
delay(min(attempt + 1, 30) * 1000L)
|
||||
true // 永远重试
|
||||
}
|
||||
.collect { message ->
|
||||
// 处理消息
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
#### 修复 ensureSessionEventSubscriptionActive:
|
||||
```kotlin
|
||||
// 之前
|
||||
val isActive = streamManager.isEventStreamActive()
|
||||
|
||||
// 现在
|
||||
val isActive = jobManager.isActive(JOB_SESSION_EVENT)
|
||||
```
|
||||
|
||||
### 2. 删除的文件
|
||||
|
||||
- `app/src/main/java/com/durian/tssparty/data/remote/StreamManager.kt`
|
||||
|
||||
### 3. 保留的 gRPC 改进
|
||||
|
||||
#### GrpcClient.kt - Keep-Alive 配置（保留）✅:
|
||||
```kotlin
|
||||
val builder = ManagedChannelBuilder
|
||||
.forAddress(host, port)
|
||||
.usePlaintext()
|
||||
.keepAliveTime(20, TimeUnit.SECONDS) // 每 20 秒 PING
|
||||
.keepAliveTimeout(5, TimeUnit.SECONDS) // 5 秒超时
|
||||
.keepAliveWithoutCalls(true) // 没有活跃 RPC 也 PING
|
||||
.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS) // 永不超时
|
||||
```
|
||||
|
||||
#### GrpcClient.kt - 网络监听（保留）✅:
|
||||
```kotlin
|
||||
fun setupNetworkMonitoring(context: Context) {
|
||||
val callback = object : ConnectivityManager.NetworkCallback() {
|
||||
override fun onAvailable(network: Network) {
|
||||
channel?.resetConnectBackoff() // 立即重连
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 架构对比
|
||||
|
||||
### 旧设计（复杂但出错）❌:
|
||||
```
|
||||
TssRepository
|
||||
├─ StreamManager (新增的抽象层)
|
||||
│ ├─ startEventStream()
|
||||
│ ├─ startMessageStream()
|
||||
│ └─ restartAllStreams()
|
||||
│
|
||||
├─ init { listen reconnection → streamManager.restartAllStreams() }
|
||||
└─ grpcClient
|
||||
```
|
||||
|
||||
### 新设计（简单可靠）✅:
|
||||
```
|
||||
TssRepository
|
||||
├─ JobManager (原有的任务管理)
|
||||
│ ├─ JOB_SESSION_EVENT → flow { subscribeSessionEvents() }.retryWhen { }
|
||||
│ ├─ JOB_MESSAGE_COLLECTION → flow { subscribeMessages() }.retryWhen { }
|
||||
│ └─ JOB_MESSAGE_SENDING → outgoingMessages.collect { }
|
||||
│
|
||||
└─ grpcClient (带 Keep-Alive + Network Monitoring)
|
||||
```
|
||||
|
||||
## 为什么新设计更好
|
||||
|
||||
1. **更少的抽象层**: 直接用 jobManager.launch，不需要 StreamManager
|
||||
2. **自动重连**: Flow.retryWhen 在流失败时自动重新发起 RPC
|
||||
3. **保持原有逻辑**: 事件处理代码保持不变，只在外面包一层 retryWhen
|
||||
4. **更好的日志**: 直接在 collect { } 里打日志，不会丢失
|
||||
5. **符合 Kotlin 风格**: Flow transformation 比 callback 更符合 Kotlin 惯用法
|
||||
|
||||
## 测试重点
|
||||
|
||||
1. ✅ 编译成功（已验证）
|
||||
2. ⏳ RegisterParty 成功（需要测试）
|
||||
3. ⏳ 事件订阅成功（看到 "Starting session event subscription" 日志）
|
||||
4. ⏳ 创建 2-of-3 会话成功
|
||||
5. ⏳ 飞行模式测试自动重连
|
||||
|
||||
## 编译结果
|
||||
|
||||
```
|
||||
BUILD SUCCESSFUL in 1m 26s
|
||||
46 actionable tasks: 17 executed, 29 up-to-date
|
||||
```
|
||||
|
||||
只有一些参数未使用的警告，没有错误。
|
||||
|
||||
## 核心教训
|
||||
|
||||
**简单就是可靠**:
|
||||
```
|
||||
工作的代码 + 官方推荐配置 + 最小改动 = 可靠的系统
|
||||
|
||||
不是:
|
||||
工作的代码 → 完全重构 → 引入新抽象 → 新问题
|
||||
```
|
||||
|
||||
**gRPC 流的正确管理方式**:
|
||||
1. 流断开时，用 Flow.retryWhen 自动重新发起 RPC（不是"恢复"）
|
||||
2. 不需要复杂的 StreamManager，Kotlin Flow 本身就是流管理器
|
||||
3. Keep-Alive 防止连接假死
|
||||
4. Network Monitoring 加速重连
|
||||
|
||||
## 下一步
|
||||
|
||||
准备测试！使用 build-install-debug.bat 安装到设备，验证：
|
||||
1. RegisterParty 是否成功
|
||||
2. 事件流是否正常工作
|
||||
3. 2-of-3 创建是否成功
|
||||
4. 网络断开重连是否自动恢复
|
||||
|
|
@ -0,0 +1,269 @@
|
|||
# 工作代码分析和正确修复方案
|
||||
|
||||
## 问题根源
|
||||
|
||||
**我犯的错误**: 在添加异常处理时，把整个流管理逻辑都改了，引入了 StreamManager 抽象层，导致流程变复杂并出现新问题。
|
||||
|
||||
**用户的要求**:
|
||||
1. ✅ 添加异常处理（如 markPartyReady 重试）
|
||||
2. ✅ 保留 gRPC 官方推荐（Keep-Alive, network monitoring）
|
||||
3. ❌ **不要改变原有的工作流程**
|
||||
|
||||
## 工作代码的逻辑（commit 41e7eed2）
|
||||
|
||||
### 1. 连接初始化序列
|
||||
|
||||
```kotlin
|
||||
// MainActivity.kt → MainViewModel.kt → TssRepository.kt
|
||||
|
||||
1. GrpcClient.connectToServer(host, port)
|
||||
↓
|
||||
2. 创建 ManagedChannel
|
||||
↓
|
||||
3. TssRepository.registerParty()
|
||||
↓
|
||||
4. grpcClient.registerParty(partyId, "temporary", "1.0.0") // 没有错误检查
|
||||
↓
|
||||
5. startSessionEventSubscription() // 立即订阅事件流
|
||||
```
|
||||
|
||||
### 2. 事件流订阅逻辑
|
||||
|
||||
```kotlin
|
||||
// TssRepository.kt - 工作的代码
|
||||
|
||||
private fun startSessionEventSubscription(subscriptionPartyId: String? = null) {
|
||||
val effectivePartyId = subscriptionPartyId ?: requirePartyId()
|
||||
currentSessionEventPartyId = effectivePartyId
|
||||
|
||||
// 关键：使用 JobManager 直接启动
|
||||
jobManager.launch(JOB_SESSION_EVENT) {
|
||||
grpcClient.subscribeSessionEvents(effectivePartyId).collect { event ->
|
||||
// 直接在这里处理事件
|
||||
when (event.eventType) {
|
||||
"session_started" -> { /* ... */ }
|
||||
"participant_joined" -> { /* ... */ }
|
||||
// ...
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
**为什么这个简单的方式可以工作?**
|
||||
- `jobManager.launch()` 自动取消同名旧 Job
|
||||
- `grpcClient.subscribeSessionEvents()` 返回一个 Flow
|
||||
- Flow 在网络断开时会自动关闭
|
||||
- 但没有自动重连机制（这是需要修复的）
|
||||
|
||||
### 3. 消息路由逻辑
|
||||
|
||||
```kotlin
|
||||
// TssRepository.kt - 消息路由
|
||||
|
||||
private fun startMessageRouting(sessionId: String, partyId: String, partyIndex: Int) {
|
||||
// 1. 启动消息收集 Job
|
||||
jobManager.launch(JOB_MESSAGE_COLLECTION) {
|
||||
subscribeToTssMessages(sessionId, partyId).collect { message ->
|
||||
tssNativeBridge.routeIncomingMessage(sessionId, message)
|
||||
}
|
||||
}
|
||||
|
||||
// 2. 同时启动消息发送 Job（在同一个 JobManager 中）
|
||||
jobManager.launch(JOB_MESSAGE_SENDING) {
|
||||
tssNativeBridge.outgoingMessages.collect { message ->
|
||||
grpcClient.sendMessage(sessionId, message)
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## 我的错误修改
|
||||
|
||||
### 错误 1: 引入了 StreamManager 抽象层
|
||||
|
||||
```kotlin
|
||||
// 新代码（错误）
|
||||
streamManager.startEventStream(
|
||||
partyId = effectivePartyId,
|
||||
onEvent = { event -> /* callback */ }
|
||||
)
|
||||
```
|
||||
|
||||
**问题**:
|
||||
- 增加了一层不必要的抽象
|
||||
- StreamManager 的实现可能有 bug
|
||||
- 日志显示 StreamManager 根本没有启动
|
||||
|
||||
### 错误 2: 修改了连接重建后的流恢复逻辑
|
||||
|
||||
```kotlin
|
||||
// 旧代码（工作的）
|
||||
grpcConnectionEvents
|
||||
.filter { it is GrpcConnectionEvent.Reconnected }
|
||||
.collect {
|
||||
onReconnectedCallback?.invoke() // 简单的 callback
|
||||
}
|
||||
|
||||
// 新代码（复杂但出错）
|
||||
grpcConnectionEvents
|
||||
.filter { it is GrpcConnectionEvent.Reconnected }
|
||||
.collect {
|
||||
streamManager.restartAllStreams() // StreamManager 可能有问题
|
||||
}
|
||||
```
|
||||
|
||||
## 正确的修复方案
|
||||
|
||||
### 保留的部分（这些是好的）✅
|
||||
|
||||
1. **gRPC Keep-Alive 配置**（GrpcClient.kt line 143-150）:
|
||||
```kotlin
|
||||
val builder = ManagedChannelBuilder
|
||||
.forAddress(host, port)
|
||||
.keepAliveTime(20, TimeUnit.SECONDS)
|
||||
.keepAliveTimeout(5, TimeUnit.SECONDS)
|
||||
.keepAliveWithoutCalls(true)
|
||||
.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS)
|
||||
```
|
||||
|
||||
2. **Android 网络监听**（GrpcClient.kt line 151-183）:
|
||||
```kotlin
|
||||
fun setupNetworkMonitoring(context: Context) {
|
||||
val callback = object : ConnectivityManager.NetworkCallback() {
|
||||
override fun onAvailable(network: Network) {
|
||||
channel?.resetConnectBackoff()
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
3. **registerParty 错误检查**（TssRepository.kt line 489-494）:
|
||||
```kotlin
|
||||
val registerResult = grpcClient.registerParty(partyId, "temporary", "1.0.0")
|
||||
if (registerResult.isFailure) {
|
||||
throw registerResult.exceptionOrNull() ?: Exception("Failed to register party")
|
||||
}
|
||||
```
|
||||
|
||||
4. **markPartyReady 重试机制**（TssRepository.kt line ~2140）:
|
||||
```kotlin
|
||||
repeat(5) { attempt ->
|
||||
if (markReadySuccess) return@repeat
|
||||
val markReadyResult = grpcClient.markPartyReady(sessionId, partyId)
|
||||
if (markReadyResult.isSuccess) {
|
||||
markReadySuccess = true
|
||||
return@repeat
|
||||
}
|
||||
delay((attempt + 1) * 500L)
|
||||
}
|
||||
```
|
||||
|
||||
### 需要回退的部分（这些破坏了原有逻辑）❌
|
||||
|
||||
1. **删除 StreamManager**:
|
||||
- 删除 `StreamManager.kt` 文件
|
||||
- 删除 TssRepository.kt 中的 `streamManager` 实例
|
||||
|
||||
2. **恢复原有的事件订阅逻辑**:
|
||||
```kotlin
|
||||
// 恢复为这样（简单直接）
|
||||
private fun startSessionEventSubscription(subscriptionPartyId: String? = null) {
|
||||
val effectivePartyId = subscriptionPartyId ?: requirePartyId()
|
||||
currentSessionEventPartyId = effectivePartyId
|
||||
|
||||
jobManager.launch(JOB_SESSION_EVENT) {
|
||||
// 添加 retryWhen 自动重连（新增的改进）
|
||||
flow {
|
||||
grpcClient.subscribeSessionEvents(effectivePartyId).collect { emit(it) }
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
Log.w(TAG, "Event stream failed (attempt ${attempt + 1}), retrying: ${cause.message}")
|
||||
delay(min(attempt + 1, 30) * 1000L)
|
||||
true // 永远重试
|
||||
}
|
||||
.collect { event ->
|
||||
// 直接处理事件（保持原有逻辑不变）
|
||||
Log.d(TAG, "=== Session event received ===")
|
||||
when (event.eventType) {
|
||||
"session_started" -> { /* ... */ }
|
||||
// ...
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
3. **删除 GrpcClient 中复杂的 reconnection callback**:
|
||||
- 保持简单的连接状态 Flow
|
||||
- 不需要复杂的 reSubscribeStreams() 逻辑
|
||||
|
||||
## 正确的架构
|
||||
|
||||
```
|
||||
简单而可靠的架构:
|
||||
|
||||
GrpcClient (基础层)
|
||||
├─ Keep-Alive 配置 ✅
|
||||
├─ Network Monitoring ✅
|
||||
├─ subscribeSessionEvents() → Flow ✅
|
||||
└─ subscribeMessages() → Flow ✅
|
||||
|
||||
TssRepository (业务层)
|
||||
├─ JobManager 管理所有协程 ✅
|
||||
├─ jobManager.launch(JOB_SESSION_EVENT) {
|
||||
│ flow { grpcClient.subscribeSessionEvents().collect { emit(it) } }
|
||||
│ .retryWhen { ... } ← 新增自动重连
|
||||
│ .collect { event -> /* 处理 */ }
|
||||
│ }
|
||||
└─ 同样的模式用于消息流 ✅
|
||||
```
|
||||
|
||||
## 实施步骤
|
||||
|
||||
### 步骤 1: 回退 TssRepository.kt 的事件订阅逻辑
|
||||
|
||||
```kotlin
|
||||
// 删除 StreamManager 相关代码（line 217-242）
|
||||
- private val streamManager = StreamManager(grpcClient, repositoryScope)
|
||||
- init { repositoryScope.launch { grpcConnectionEvents... streamManager.restartAllStreams() } }
|
||||
|
||||
// 恢复 startSessionEventSubscription 为原来的简单版本（line 511-612）
|
||||
// 但在 collect 外包一层 flow { }.retryWhen { }
|
||||
```
|
||||
|
||||
### 步骤 2: 删除 StreamManager.kt 文件
|
||||
|
||||
```bash
|
||||
rm StreamManager.kt
|
||||
```
|
||||
|
||||
### 步骤 3: 简化 GrpcClient.kt 的重连逻辑
|
||||
|
||||
```kotlin
|
||||
// 删除复杂的 reSubscribeStreams() 方法
|
||||
// 保留简单的 GrpcConnectionEvent 发送
|
||||
```
|
||||
|
||||
### 步骤 4: 测试验证
|
||||
|
||||
1. 编译成功
|
||||
2. 启动时 RegisterParty 成功
|
||||
3. 事件订阅成功（看到 "Starting session event subscription" 日志）
|
||||
4. 创建 2-of-3 会话成功
|
||||
5. 飞行模式测试自动重连
|
||||
|
||||
## 总结
|
||||
|
||||
**核心教训**:
|
||||
- ❌ 不要过度设计（StreamManager 是不必要的抽象）
|
||||
- ✅ 在原有工作的代码基础上做最小改动
|
||||
- ✅ 保留 gRPC 官方推荐的配置（Keep-Alive, network monitoring）
|
||||
- ✅ 只在必要的地方添加错误处理和重试逻辑
|
||||
|
||||
**修复原则**:
|
||||
```
|
||||
旧代码 + 官方推荐 + 最小改动 = 可靠的解决方案
|
||||
|
||||
不是: 旧代码 → 完全重构 → StreamManager → 新问题
|
||||
```
|
||||
|
|
@ -1,282 +0,0 @@
|
|||
package com.durian.tssparty.data.remote
|
||||
|
||||
import android.util.Log
|
||||
import kotlinx.coroutines.*
|
||||
import kotlinx.coroutines.flow.*
|
||||
import kotlin.math.min
|
||||
|
||||
/**
|
||||
* Manages gRPC stream lifecycle with automatic reconnection
|
||||
*
|
||||
* Key principles:
|
||||
* 1. Streams cannot be "restored" - they must be re-initiated after disconnection
|
||||
* 2. Each stream maintains its configuration for automatic restart
|
||||
* 3. Exponential backoff for failed stream attempts
|
||||
* 4. All streams are supervised - failure of one doesn't affect others
|
||||
*/
|
||||
class StreamManager(
|
||||
private val grpcClient: GrpcClient,
|
||||
private val scope: CoroutineScope
|
||||
) {
|
||||
companion object {
|
||||
private const val TAG = "StreamManager"
|
||||
private const val MAX_RETRY_DELAY_SECONDS = 30L
|
||||
}
|
||||
|
||||
// Active stream configurations
|
||||
private var eventStreamConfig: EventStreamConfig? = null
|
||||
private var messageStreamConfig: MessageStreamConfig? = null
|
||||
|
||||
// Active stream jobs
|
||||
private var eventStreamJob: Job? = null
|
||||
private var messageStreamJob: Job? = null
|
||||
|
||||
// Supervision flags
|
||||
private var shouldMaintainEventStream = false
|
||||
private var shouldMaintainMessageStream = false
|
||||
|
||||
/**
|
||||
* Start event stream with automatic reconnection
|
||||
*
|
||||
* @param partyId Party ID to subscribe events for
|
||||
* @param onEvent Callback for each event received
|
||||
* @param onError Callback for stream errors (before retry)
|
||||
*/
|
||||
fun startEventStream(
|
||||
partyId: String,
|
||||
onEvent: suspend (SessionEventData) -> Unit,
|
||||
onError: ((Throwable) -> Unit)? = null
|
||||
) {
|
||||
Log.d(TAG, "Starting event stream for partyId=$partyId")
|
||||
|
||||
// Save configuration
|
||||
eventStreamConfig = EventStreamConfig(partyId, onEvent, onError)
|
||||
shouldMaintainEventStream = true
|
||||
|
||||
// Stop existing stream if any
|
||||
eventStreamJob?.cancel()
|
||||
|
||||
// Launch new stream with retry logic
|
||||
eventStreamJob = launchEventStream(partyId, onEvent, onError)
|
||||
}
|
||||
|
||||
/**
|
||||
* Start message stream with automatic reconnection
|
||||
*
|
||||
* @param sessionId Session ID
|
||||
* @param partyId Party ID
|
||||
* @param partyIndex Party index in the session
|
||||
* @param onMessage Callback for each message received
|
||||
* @param onError Callback for stream errors (before retry)
|
||||
*/
|
||||
fun startMessageStream(
|
||||
sessionId: String,
|
||||
partyId: String,
|
||||
partyIndex: Int,
|
||||
onMessage: suspend (IncomingMessage) -> Unit,
|
||||
onError: ((Throwable) -> Unit)? = null
|
||||
) {
|
||||
Log.d(TAG, "Starting message stream for sessionId=$sessionId, partyId=$partyId")
|
||||
|
||||
// Save configuration
|
||||
messageStreamConfig = MessageStreamConfig(sessionId, partyId, partyIndex, onMessage, onError)
|
||||
shouldMaintainMessageStream = true
|
||||
|
||||
// Stop existing stream if any
|
||||
messageStreamJob?.cancel()
|
||||
|
||||
// Launch new stream with retry logic
|
||||
messageStreamJob = launchMessageStream(sessionId, partyId, partyIndex, onMessage, onError)
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop event stream
|
||||
*/
|
||||
fun stopEventStream() {
|
||||
Log.d(TAG, "Stopping event stream")
|
||||
shouldMaintainEventStream = false
|
||||
eventStreamConfig = null
|
||||
eventStreamJob?.cancel()
|
||||
eventStreamJob = null
|
||||
}
|
||||
|
||||
/**
|
||||
* Stop message stream
|
||||
*/
|
||||
fun stopMessageStream() {
|
||||
Log.d(TAG, "Stopping message stream")
|
||||
shouldMaintainMessageStream = false
|
||||
messageStreamConfig = null
|
||||
messageStreamJob?.cancel()
|
||||
messageStreamJob = null
|
||||
}
|
||||
|
||||
/**
|
||||
* Restart all active streams (called after reconnection)
|
||||
*/
|
||||
fun restartAllStreams() {
|
||||
Log.d(TAG, "Restarting all active streams")
|
||||
|
||||
// Restart event stream if it was active
|
||||
if (shouldMaintainEventStream) {
|
||||
eventStreamConfig?.let { config ->
|
||||
Log.d(TAG, "Restarting event stream for partyId=${config.partyId}")
|
||||
eventStreamJob?.cancel()
|
||||
eventStreamJob = launchEventStream(config.partyId, config.onEvent, config.onError)
|
||||
}
|
||||
}
|
||||
|
||||
// Restart message stream if it was active
|
||||
if (shouldMaintainMessageStream) {
|
||||
messageStreamConfig?.let { config ->
|
||||
Log.d(TAG, "Restarting message stream for sessionId=${config.sessionId}")
|
||||
messageStreamJob?.cancel()
|
||||
messageStreamJob = launchMessageStream(
|
||||
config.sessionId,
|
||||
config.partyId,
|
||||
config.partyIndex,
|
||||
config.onMessage,
|
||||
config.onError
|
||||
)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Check if event stream is active
|
||||
*/
|
||||
fun isEventStreamActive(): Boolean = shouldMaintainEventStream
|
||||
|
||||
/**
|
||||
* Check if message stream is active
|
||||
*/
|
||||
fun isMessageStreamActive(): Boolean = shouldMaintainMessageStream
|
||||
|
||||
/**
|
||||
* Get current event stream party ID
|
||||
*/
|
||||
fun getEventStreamPartyId(): String? = eventStreamConfig?.partyId
|
||||
|
||||
/**
|
||||
* Get current message stream session ID
|
||||
*/
|
||||
fun getMessageStreamSessionId(): String? = messageStreamConfig?.sessionId
|
||||
|
||||
// ===== Private Implementation =====
|
||||
|
||||
/**
|
||||
* Launch event stream with exponential backoff retry
|
||||
*/
|
||||
private fun launchEventStream(
|
||||
partyId: String,
|
||||
onEvent: suspend (SessionEventData) -> Unit,
|
||||
onError: ((Throwable) -> Unit)?
|
||||
): Job = scope.launch {
|
||||
Log.d(TAG, "Launching event stream flow with auto-retry for partyId=$partyId")
|
||||
|
||||
flow {
|
||||
// Re-initiate gRPC subscription (not "restore")
|
||||
grpcClient.subscribeSessionEvents(partyId).collect { event ->
|
||||
emit(event)
|
||||
}
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
// Stop retrying if stream was explicitly stopped
|
||||
if (!shouldMaintainEventStream) {
|
||||
Log.d(TAG, "Event stream stopped, not retrying")
|
||||
return@retryWhen false
|
||||
}
|
||||
|
||||
// Calculate exponential backoff delay
|
||||
val delaySeconds = min(attempt + 1, MAX_RETRY_DELAY_SECONDS)
|
||||
Log.w(TAG, "Event stream failed (attempt ${attempt + 1}), retrying in ${delaySeconds}s: ${cause.message}")
|
||||
|
||||
// Notify error callback
|
||||
onError?.invoke(cause)
|
||||
|
||||
// Wait before retry
|
||||
delay(delaySeconds * 1000)
|
||||
true // Always retry if stream should be maintained
|
||||
}
|
||||
.catch { e ->
|
||||
// This should not happen due to retryWhen, but handle just in case
|
||||
Log.e(TAG, "Event stream terminated unexpectedly", e)
|
||||
onError?.invoke(e)
|
||||
}
|
||||
.collect { event ->
|
||||
try {
|
||||
onEvent(event)
|
||||
} catch (e: Exception) {
|
||||
Log.e(TAG, "Error processing event", e)
|
||||
// Don't let handler errors kill the stream
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Launch message stream with exponential backoff retry
|
||||
*/
|
||||
private fun launchMessageStream(
|
||||
sessionId: String,
|
||||
partyId: String,
|
||||
partyIndex: Int,
|
||||
onMessage: suspend (IncomingMessage) -> Unit,
|
||||
onError: ((Throwable) -> Unit)?
|
||||
): Job = scope.launch {
|
||||
Log.d(TAG, "Launching message stream flow with auto-retry for sessionId=$sessionId")
|
||||
|
||||
flow {
|
||||
// Re-initiate gRPC subscription (not "restore")
|
||||
grpcClient.subscribeMessages(sessionId, partyId).collect { message ->
|
||||
emit(message)
|
||||
}
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
// Stop retrying if stream was explicitly stopped
|
||||
if (!shouldMaintainMessageStream) {
|
||||
Log.d(TAG, "Message stream stopped, not retrying")
|
||||
return@retryWhen false
|
||||
}
|
||||
|
||||
// Calculate exponential backoff delay
|
||||
val delaySeconds = min(attempt + 1, MAX_RETRY_DELAY_SECONDS)
|
||||
Log.w(TAG, "Message stream failed (attempt ${attempt + 1}), retrying in ${delaySeconds}s: ${cause.message}")
|
||||
|
||||
// Notify error callback
|
||||
onError?.invoke(cause)
|
||||
|
||||
// Wait before retry
|
||||
delay(delaySeconds * 1000)
|
||||
true // Always retry if stream should be maintained
|
||||
}
|
||||
.catch { e ->
|
||||
// This should not happen due to retryWhen, but handle just in case
|
||||
Log.e(TAG, "Message stream terminated unexpectedly", e)
|
||||
onError?.invoke(e)
|
||||
}
|
||||
.collect { message ->
|
||||
try {
|
||||
onMessage(message)
|
||||
} catch (e: Exception) {
|
||||
Log.e(TAG, "Error processing message", e)
|
||||
// Don't let handler errors kill the stream
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// ===== Configuration Data Classes =====
|
||||
|
||||
private data class EventStreamConfig(
|
||||
val partyId: String,
|
||||
val onEvent: suspend (SessionEventData) -> Unit,
|
||||
val onError: ((Throwable) -> Unit)?
|
||||
)
|
||||
|
||||
private data class MessageStreamConfig(
|
||||
val sessionId: String,
|
||||
val partyId: String,
|
||||
val partyIndex: Int,
|
||||
val onMessage: suspend (IncomingMessage) -> Unit,
|
||||
val onError: ((Throwable) -> Unit)?
|
||||
)
|
||||
}
|
||||
|
|
@ -14,7 +14,6 @@ import com.durian.tssparty.data.remote.GrpcConnectionState
|
|||
import com.durian.tssparty.data.remote.IncomingMessage
|
||||
import com.durian.tssparty.data.remote.JoinSessionData
|
||||
import com.durian.tssparty.data.remote.SessionEventData
|
||||
import com.durian.tssparty.data.remote.StreamManager
|
||||
import com.durian.tssparty.domain.model.*
|
||||
import com.durian.tssparty.util.AddressUtils
|
||||
import com.durian.tssparty.util.TransactionUtils
|
||||
|
|
@ -214,36 +213,11 @@ class TssRepository @Inject constructor(
|
|||
coroutineExceptionHandler
|
||||
)
|
||||
|
||||
/**
|
||||
* StreamManager - 管理 gRPC 双向流的生命周期
|
||||
*
|
||||
* 【架构重构 - 可靠的流管理机制】
|
||||
*
|
||||
* 核心原则（来自 gRPC 官方文档）：
|
||||
* - gRPC 流无法"恢复"，必须重新发起 RPC 调用
|
||||
* - 流断开后需要重新调用 subscribeSessionEvents() / subscribeMessages()
|
||||
* - 保存流配置，自动重试失败的流（指数退避）
|
||||
*
|
||||
* 修复的关键问题：
|
||||
* 1. 旧设计尝试"恢复"已关闭的 Flow → 失败（Flow 不是持久化对象）
|
||||
* 2. 网络断开后 eventStreamSubscribed flag 被清除 → callback 不触发
|
||||
* 3. 流失败后没有自动重试机制 → 永久失败
|
||||
*
|
||||
* StreamManager 功能：
|
||||
* - 保存每个流的配置（partyId, sessionId 等）
|
||||
* - 流失败后自动重新发起 RPC（不是"恢复"）
|
||||
* - 使用 Flow.retryWhen 实现指数退避重试
|
||||
* - 网络重连时，重启所有活跃的流
|
||||
*
|
||||
* 参考资料：
|
||||
* - https://github.com/grpc/grpc-java/issues/8177
|
||||
* - https://grpc.io/docs/guides/keepalive/
|
||||
*/
|
||||
private val streamManager = StreamManager(grpcClient, repositoryScope)
|
||||
|
||||
companion object {
|
||||
// Job 名称常量
|
||||
private const val JOB_MESSAGE_COLLECTION = "message_collection"
|
||||
private const val JOB_MESSAGE_SENDING = "message_sending"
|
||||
private const val JOB_SESSION_EVENT = "session_event"
|
||||
private const val JOB_SESSION_STATUS_POLLING = "session_status_polling"
|
||||
private const val JOB_PROGRESS_COLLECTION = "progress_collection"
|
||||
|
|
@ -316,19 +290,6 @@ class TssRepository @Inject constructor(
|
|||
// Account service URL (configurable via settings)
|
||||
private var accountServiceUrl: String = "https://rwaapi.szaiai.com"
|
||||
|
||||
init {
|
||||
// Monitor gRPC reconnection events and restart streams
|
||||
// IMPORTANT: Don't use callback pattern - use event Flow for better reliability
|
||||
repositoryScope.launch {
|
||||
grpcConnectionEvents
|
||||
.filter { it is GrpcConnectionEvent.Reconnected }
|
||||
.collect {
|
||||
android.util.Log.d("TssRepository", "gRPC reconnected, restarting streams via StreamManager...")
|
||||
// StreamManager will re-initiate RPC calls (not "restore" closed Flows)
|
||||
streamManager.restartAllStreams()
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* HTTP client for API calls
|
||||
|
|
@ -485,7 +446,15 @@ class TssRepository @Inject constructor(
|
|||
newPartyId
|
||||
}
|
||||
|
||||
grpcClient.registerParty(partyId, "temporary", "1.0.0")
|
||||
// Register with gRPC and check result
|
||||
val registerResult = grpcClient.registerParty(partyId, "temporary", "1.0.0")
|
||||
if (registerResult.isFailure) {
|
||||
val error = registerResult.exceptionOrNull()
|
||||
android.util.Log.e("TssRepository", "Failed to register party: ${error?.message}")
|
||||
throw error ?: Exception("Failed to register party")
|
||||
}
|
||||
|
||||
android.util.Log.d("TssRepository", "Party registered successfully: $partyId")
|
||||
|
||||
// Subscribe to session events immediately after registration (like Electron does)
|
||||
startSessionEventSubscription()
|
||||
|
|
@ -507,10 +476,20 @@ class TssRepository @Inject constructor(
|
|||
currentSessionEventPartyId = effectivePartyId
|
||||
android.util.Log.d("TssRepository", "Starting session event subscription for partyId: $effectivePartyId (device partyId: $devicePartyId)")
|
||||
|
||||
// Use StreamManager for reliable stream management with auto-reconnection
|
||||
streamManager.startEventStream(
|
||||
partyId = effectivePartyId,
|
||||
onEvent = { event ->
|
||||
// 使用 JobManager 启动（自动取消同名旧 Job）
|
||||
// 添加 Flow.retryWhen 实现自动重连（基于 gRPC 官方推荐）
|
||||
jobManager.launch(JOB_SESSION_EVENT) {
|
||||
flow {
|
||||
grpcClient.subscribeSessionEvents(effectivePartyId).collect { event ->
|
||||
emit(event)
|
||||
}
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
android.util.Log.w("TssRepository", "Event stream failed (attempt ${attempt + 1}), retrying in ${kotlin.math.min(attempt + 1, 30)}s: ${cause.message}")
|
||||
delay(kotlin.math.min(attempt + 1, 30) * 1000L) // 指数退避，最多 30 秒
|
||||
true // 永远重试
|
||||
}
|
||||
.collect { event ->
|
||||
android.util.Log.d("TssRepository", "=== Session event received ===")
|
||||
android.util.Log.d("TssRepository", " eventType: ${event.eventType}")
|
||||
android.util.Log.d("TssRepository", " sessionId: ${event.sessionId}")
|
||||
|
|
@ -595,12 +574,8 @@ class TssRepository @Inject constructor(
|
|||
android.util.Log.d("TssRepository", " Reason: sessionId mismatch (event: ${event.sessionId}, active: ${activeSession.sessionId})")
|
||||
}
|
||||
}
|
||||
},
|
||||
onError = { error ->
|
||||
android.util.Log.e("TssRepository", "Event stream error: ${error.message}")
|
||||
// StreamManager will automatically retry
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -614,8 +589,8 @@ class TssRepository @Inject constructor(
|
|||
* partyId from keygen (shareEntity.partyId).
|
||||
*/
|
||||
private fun ensureSessionEventSubscriptionActive(signingPartyId: String? = null) {
|
||||
// Check if the event stream is still active using StreamManager
|
||||
val isActive = streamManager.isEventStreamActive()
|
||||
// Check if the event stream Job is still active using JobManager
|
||||
val isActive = jobManager.isActive(JOB_SESSION_EVENT)
|
||||
val devicePartyId = requirePartyId() // Ensure partyId is initialized
|
||||
val effectivePartyId = signingPartyId ?: currentSessionEventPartyId ?: devicePartyId
|
||||
android.util.Log.d("TssRepository", "Checking session event subscription: isActive=$isActive, effectivePartyId=$effectivePartyId")
|
||||
|
|
@ -2068,8 +2043,7 @@ class TssRepository @Inject constructor(
|
|||
android.util.Log.d("TssRepository", "Starting message routing: sessionId=$sessionId, routingPartyId=$effectivePartyId")
|
||||
|
||||
// Part 1: Collect outgoing messages from TSS and route via gRPC
|
||||
// This doesn't need StreamManager - it's a local Flow collection
|
||||
jobManager.launch(JOB_MESSAGE_COLLECTION) {
|
||||
jobManager.launch(JOB_MESSAGE_SENDING) {
|
||||
tssNativeBridge.outgoingMessages.collect { message ->
|
||||
val payload = Base64.decode(message.payload, Base64.NO_WRAP)
|
||||
grpcClient.routeMessage(
|
||||
|
|
@ -2084,12 +2058,19 @@ class TssRepository @Inject constructor(
|
|||
}
|
||||
|
||||
// Part 2: Subscribe to incoming messages from gRPC and send to TSS
|
||||
// Use StreamManager for reliable gRPC stream management with auto-reconnection
|
||||
streamManager.startMessageStream(
|
||||
sessionId = sessionId,
|
||||
partyId = effectivePartyId,
|
||||
partyIndex = partyIndex,
|
||||
onMessage = { message ->
|
||||
// 添加 Flow.retryWhen 实现自动重连（基于 gRPC 官方推荐）
|
||||
jobManager.launch(JOB_MESSAGE_COLLECTION) {
|
||||
flow {
|
||||
grpcClient.subscribeMessages(sessionId, effectivePartyId).collect { message ->
|
||||
emit(message)
|
||||
}
|
||||
}
|
||||
.retryWhen { cause, attempt ->
|
||||
android.util.Log.w("TssRepository", "Message stream failed (attempt ${attempt + 1}), retrying in ${kotlin.math.min(attempt + 1, 30)}s: ${cause.message}")
|
||||
delay(kotlin.math.min(attempt + 1, 30) * 1000L) // 指数退避，最多 30 秒
|
||||
true // 永远重试
|
||||
}
|
||||
.collect { message ->
|
||||
// Find party index from party ID
|
||||
val session = _currentSession.value
|
||||
val fromPartyIndex = session?.participants?.find { it.partyId == message.fromParty }?.partyIndex
|
||||
|
|
@ -2103,12 +2084,8 @@ class TssRepository @Inject constructor(
|
|||
} else {
|
||||
android.util.Log.w("TssRepository", "Unknown fromParty: ${message.fromParty}, skipping message")
|
||||
}
|
||||
},
|
||||
onError = { error ->
|
||||
android.util.Log.e("TssRepository", "Message stream error: ${error.message}")
|
||||
// StreamManager will automatically retry
|
||||
}
|
||||
)
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
|
|||
Loading…
Reference in New Issue