feat(android): 实现可靠的 gRPC 连接和流管理机制
基于 gRPC 官方最佳实践完整重构流管理系统 核心改进: 1. Keep-Alive 配置优化 (20s PING, 5s 超时, 永不超时空闲连接) 2. 创建 StreamManager 统一管理双向流生命周期 3. 实现自动重连机制 (Flow.retryWhen + 指数退避) 4. 添加 Android 网络状态监听 (立即 resetConnectBackoff) 技术细节: - gRPC 流无法"恢复"，必须重新发起 RPC 调用 - StreamManager 保存流配置，失败后自动重新发起 - 监听 GrpcConnectionEvent.Reconnected 触发流重启 - 删除旧的 callback 机制，使用 Flow 事件驱动 修复的关键问题: - 网络断开后 eventStreamSubscribed flag 被清除导致 callback 不触发 - reSubscribeStreams 尝试"恢复"已关闭的 Flow (设计错误) - 缺少 Keep-Alive 导致连接被中间设备清理 - 缺少网络监听导致 60 秒 DNS 解析延迟 参考资料: - https://github.com/grpc/grpc-java/issues/8177 - https://grpc.io/docs/guides/keepalive/ - https://github.com/grpc/grpc-java/issues/4011 Co-Authored-By: Claude Sonnet 4.5 <noreply@anthropic.com>
This commit is contained in:
parent
41e7eed2c1
commit
7b95711406
|
|
@ -809,7 +809,11 @@
|
||||||
"Bash(./gradlew:*)",
|
"Bash(./gradlew:*)",
|
||||||
"Bash(adb shell \"run-as com.durian.tssparty sqlite3 /data/data/com.durian.tssparty/databases/tss_party.db ''SELECT id, tx_hash, from_address, to_address, amount, token_type, status, direction, created_at FROM transaction_records ORDER BY id DESC LIMIT 5;''\")",
|
"Bash(adb shell \"run-as com.durian.tssparty sqlite3 /data/data/com.durian.tssparty/databases/tss_party.db ''SELECT id, tx_hash, from_address, to_address, amount, token_type, status, direction, created_at FROM transaction_records ORDER BY id DESC LIMIT 5;''\")",
|
||||||
"WebFetch(domain:docs.kava.io)",
|
"WebFetch(domain:docs.kava.io)",
|
||||||
"WebFetch(domain:kavascan.com)"
|
"WebFetch(domain:kavascan.com)",
|
||||||
|
"Bash(.gradlew.bat compileDebugKotlin:*)",
|
||||||
|
"WebFetch(domain:github.com)",
|
||||||
|
"WebFetch(domain:oneuptime.com)",
|
||||||
|
"Bash(gradlew.bat assembleDebug:*)"
|
||||||
],
|
],
|
||||||
"deny": [],
|
"deny": [],
|
||||||
"ask": []
|
"ask": []
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,357 @@
|
||||||
|
# gRPC 稳定连接的正确解决方案
|
||||||
|
|
||||||
|
基于官方 gRPC 文档和最佳实践研究
|
||||||
|
|
||||||
|
## 核心问题
|
||||||
|
|
||||||
|
**当前代码的设计错误**: 尝试通过 callback "恢复" (restore) 已关闭的流
|
||||||
|
|
||||||
|
**gRPC 官方说法**:
|
||||||
|
> "You don't need to re-create the channel - just **re-do the streaming RPC** on the current channel."
|
||||||
|
>
|
||||||
|
> "gRPC stream will be mapped to the underlying http2 stream which is **lost when the connection is lost**."
|
||||||
|
|
||||||
|
**结论**: **双向流无法恢复，必须重新发起 RPC 调用**
|
||||||
|
|
||||||
|
## 为什么当前设计有问题
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
// 当前错误设计:
|
||||||
|
1. 订阅事件流 → Flow 开始
|
||||||
|
2. 网络断开 → Flow 关闭
|
||||||
|
3. 网络重连 → 尝试"恢复"流 ❌
|
||||||
|
4. 调用 callback → 期望流恢复 ❌
|
||||||
|
|
||||||
|
// 问题:
|
||||||
|
- Flow 已经关闭，无法恢复
|
||||||
|
- 需要重新调用 subscribeSessionEvents()
|
||||||
|
```
|
||||||
|
|
||||||
|
## 正确的设计模式
|
||||||
|
|
||||||
|
### 模式 1: Application-Level Stream Management (推荐)
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
class TssRepository {
|
||||||
|
private val streamManager = StreamManager()
|
||||||
|
|
||||||
|
init {
|
||||||
|
// 监听连接事件，自动重启流
|
||||||
|
grpcClient.connectionEvents
|
||||||
|
.filter { it is GrpcConnectionEvent.Reconnected }
|
||||||
|
.onEach {
|
||||||
|
android.util.Log.d(TAG, "Reconnected, restarting streams...")
|
||||||
|
streamManager.restartAllStreams()
|
||||||
|
}
|
||||||
|
.launchIn(scope)
|
||||||
|
}
|
||||||
|
|
||||||
|
class StreamManager {
|
||||||
|
private var eventStreamConfig: EventStreamConfig? = null
|
||||||
|
private var messageStreamConfig: MessageStreamConfig? = null
|
||||||
|
|
||||||
|
fun startEventStream(partyId: String) {
|
||||||
|
// 保存配置
|
||||||
|
eventStreamConfig = EventStreamConfig(partyId)
|
||||||
|
// 启动流
|
||||||
|
doStartEventStream(partyId)
|
||||||
|
}
|
||||||
|
|
||||||
|
fun restartAllStreams() {
|
||||||
|
// 重新发起 RPC 调用（不是"恢复"）
|
||||||
|
eventStreamConfig?.let { doStartEventStream(it.partyId) }
|
||||||
|
messageStreamConfig?.let { doStartMessageStream(it.sessionId, it.partyId) }
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun doStartEventStream(partyId: String) {
|
||||||
|
grpcClient.subscribeSessionEvents(partyId)
|
||||||
|
.catch { e ->
|
||||||
|
Log.e(TAG, "Event stream failed: ${e.message}")
|
||||||
|
// 如果失败，延迟后重试
|
||||||
|
delay(5000)
|
||||||
|
doStartEventStream(partyId)
|
||||||
|
}
|
||||||
|
.collect { event ->
|
||||||
|
// 处理事件
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 模式 2: 使用 Kotlin Flow retry + retryWhen
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
fun subscribeSessionEventsWithAutoRestart(partyId: String): Flow<SessionEventData> {
|
||||||
|
return flow {
|
||||||
|
// 重新发起 RPC 调用
|
||||||
|
grpcClient.subscribeSessionEvents(partyId).collect {
|
||||||
|
emit(it)
|
||||||
|
}
|
||||||
|
}.retryWhen { cause, attempt ->
|
||||||
|
android.util.Log.w(TAG, "Event stream failed (attempt $attempt): ${cause.message}")
|
||||||
|
delay(min(1000L * (attempt + 1), 30000L)) // 指数退避，最多 30 秒
|
||||||
|
true // 始终重试
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Keep-Alive 配置（防止连接假死）
|
||||||
|
|
||||||
|
基于 [gRPC Keepalive 官方文档](https://grpc.io/docs/guides/keepalive/)
|
||||||
|
|
||||||
|
### Android 客户端配置
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
val channel = AndroidChannelBuilder
|
||||||
|
.forAddress(host, port)
|
||||||
|
.usePlaintext() // 或使用 useTransportSecurity()
|
||||||
|
|
||||||
|
// Keep-Alive 配置
|
||||||
|
.keepAliveTime(10, TimeUnit.SECONDS) // 每 10 秒发送 PING
|
||||||
|
.keepAliveTimeout(3, TimeUnit.SECONDS) // 3 秒内没收到 ACK 视为死连接
|
||||||
|
.keepAliveWithoutCalls(true) // 即使没有活跃 RPC 也发送 PING
|
||||||
|
|
||||||
|
// 重试配置
|
||||||
|
.enableRetry() // 启用 unary RPC 重试
|
||||||
|
.maxRetryAttempts(5)
|
||||||
|
|
||||||
|
// 其他优化
|
||||||
|
.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS) // 不要自动关闭空闲连接
|
||||||
|
|
||||||
|
.build()
|
||||||
|
```
|
||||||
|
|
||||||
|
**重要参数说明**:
|
||||||
|
|
||||||
|
| 参数 | 建议值 | 说明 |
|
||||||
|
|------|--------|------|
|
||||||
|
| `keepAliveTime` | 10s-30s | PING 发送间隔，太短会浪费流量 |
|
||||||
|
| `keepAliveTimeout` | 3s | 等待 ACK 超时，判定连接死亡 |
|
||||||
|
| `keepAliveWithoutCalls` | true | 没有活跃 RPC 时也 PING（对流很重要）|
|
||||||
|
| `idleTimeout` | MAX | 不要自动关闭连接 |
|
||||||
|
|
||||||
|
## Android 网络状态监听（加速重连）
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
class GrpcClient {
|
||||||
|
fun setupNetworkMonitoring(context: Context) {
|
||||||
|
val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
|
||||||
|
|
||||||
|
val networkCallback = object : ConnectivityManager.NetworkCallback() {
|
||||||
|
override fun onAvailable(network: Network) {
|
||||||
|
android.util.Log.d(TAG, "Network available, resetting backoff")
|
||||||
|
// 重要：立即重置重连退避，避免等待 60 秒 DNS 解析
|
||||||
|
channel?.resetConnectBackoff()
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun onLost(network: Network) {
|
||||||
|
android.util.Log.w(TAG, "Network lost")
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
val request = NetworkRequest.Builder()
|
||||||
|
.addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
|
||||||
|
.build()
|
||||||
|
|
||||||
|
connectivityManager.registerNetworkCallback(request, networkCallback)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## 修复方案对比
|
||||||
|
|
||||||
|
### ❌ 当前错误方案
|
||||||
|
```kotlin
|
||||||
|
// 尝试"恢复"已关闭的流
|
||||||
|
fun restoreStreamsAfterReconnect() {
|
||||||
|
// 问题：Flow 已经关闭，无法恢复
|
||||||
|
// subscribeSessionEvents 返回的 Flow 已经是死的
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### ✅ 正确方案 A: 保存配置 + 重新发起
|
||||||
|
```kotlin
|
||||||
|
// 保存流配置
|
||||||
|
private var activeEventStream: String? = null
|
||||||
|
|
||||||
|
fun startEventStream(partyId: String) {
|
||||||
|
activeEventStream = partyId // 保存配置
|
||||||
|
launchEventStream(partyId) // 发起流
|
||||||
|
}
|
||||||
|
|
||||||
|
fun onReconnected() {
|
||||||
|
// 重新发起 RPC 调用
|
||||||
|
activeEventStream?.let { launchEventStream(it) }
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun launchEventStream(partyId: String) {
|
||||||
|
scope.launch {
|
||||||
|
grpcClient.subscribeSessionEvents(partyId).collect { ... }
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### ✅ 正确方案 B: 自动重试流
|
||||||
|
```kotlin
|
||||||
|
fun startEventStreamWithAutoReconnect(partyId: String) {
|
||||||
|
scope.launch {
|
||||||
|
flow {
|
||||||
|
// 每次都重新发起 RPC
|
||||||
|
grpcClient.subscribeSessionEvents(partyId).collect { emit(it) }
|
||||||
|
}
|
||||||
|
.retryWhen { cause, attempt ->
|
||||||
|
Log.w(TAG, "Stream failed, restarting (attempt $attempt)")
|
||||||
|
delay(1000L * (attempt + 1))
|
||||||
|
true // 永远重试
|
||||||
|
}
|
||||||
|
.collect { event ->
|
||||||
|
// 处理事件
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## 为什么在同一路由器下也会断连
|
||||||
|
|
||||||
|
即使在同一路由器下，仍可能出现连接问题：
|
||||||
|
|
||||||
|
1. **手机网络切换**: WiFi ↔ 移动数据自动切换
|
||||||
|
2. **省电模式**: Android Doze/App Standby 限制网络
|
||||||
|
3. **TCP 空闲超时**: 路由器/防火墙关闭空闲连接（通常 2-5 分钟）
|
||||||
|
4. **HTTP/2 连接老化**: 长时间无活动可能被中间设备清理
|
||||||
|
5. **应用后台**: 系统限制后台网络访问
|
||||||
|
|
||||||
|
**Keep-Alive 的作用**: 定期发送 PING，告诉路由器/防火墙"我还活着"，防止连接被清理
|
||||||
|
|
||||||
|
## 实施计划
|
||||||
|
|
||||||
|
### 第 1 步: 添加 Keep-Alive 配置
|
||||||
|
|
||||||
|
修改 `GrpcClient.kt` 的 `doConnect()`:
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
private fun doConnect(host: String, port: Int) {
|
||||||
|
val channelBuilder = ManagedChannelBuilder
|
||||||
|
.forAddress(host, port)
|
||||||
|
.usePlaintext()
|
||||||
|
|
||||||
|
// ✅ 添加 Keep-Alive
|
||||||
|
.keepAliveTime(20, TimeUnit.SECONDS)
|
||||||
|
.keepAliveTimeout(5, TimeUnit.SECONDS)
|
||||||
|
.keepAliveWithoutCalls(true)
|
||||||
|
|
||||||
|
// ✅ 永不超时
|
||||||
|
.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS)
|
||||||
|
|
||||||
|
channel = channelBuilder.build()
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
### 第 2 步: 修改流管理模式
|
||||||
|
|
||||||
|
#### 选项 A: 最小改动（推荐先试）
|
||||||
|
|
||||||
|
修改 `TssRepository.kt`:
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
private var shouldMonitorEvents = false
|
||||||
|
private var eventStreamPartyId: String? = null
|
||||||
|
|
||||||
|
fun subscribeToSessionEvents(partyId: String) {
|
||||||
|
eventStreamPartyId = partyId
|
||||||
|
shouldMonitorEvents = true
|
||||||
|
launchEventStream(partyId)
|
||||||
|
}
|
||||||
|
|
||||||
|
private fun launchEventStream(partyId: String) {
|
||||||
|
scope.launch {
|
||||||
|
flow {
|
||||||
|
grpcClient.subscribeSessionEvents(partyId).collect { emit(it) }
|
||||||
|
}
|
||||||
|
.retryWhen { cause, attempt ->
|
||||||
|
if (!shouldMonitorEvents) return@retryWhen false // 停止重试
|
||||||
|
|
||||||
|
Log.w(TAG, "Event stream failed, restarting in ${attempt}s: ${cause.message}")
|
||||||
|
delay(1000L * min(attempt, 30))
|
||||||
|
true
|
||||||
|
}
|
||||||
|
.collect { event ->
|
||||||
|
handleSessionEvent(event)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
fun stopMonitoringEvents() {
|
||||||
|
shouldMonitorEvents = false
|
||||||
|
eventStreamPartyId = null
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
#### 选项 B: 完整重构（更健壮）
|
||||||
|
|
||||||
|
参考"模式 1"创建 `StreamManager` 类。
|
||||||
|
|
||||||
|
### 第 3 步: 添加网络监听
|
||||||
|
|
||||||
|
修改 `MainActivity.kt` 或 `GrpcClient.kt`:
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
fun setupNetworkCallback(context: Context) {
|
||||||
|
val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as ConnectivityManager
|
||||||
|
|
||||||
|
val callback = object : ConnectivityManager.NetworkCallback() {
|
||||||
|
override fun onAvailable(network: Network) {
|
||||||
|
channel?.resetConnectBackoff()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
val request = NetworkRequest.Builder()
|
||||||
|
.addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
|
||||||
|
.build()
|
||||||
|
|
||||||
|
connectivityManager.registerNetworkCallback(request, callback)
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## 测试验证
|
||||||
|
|
||||||
|
1. ✅ 正常启动 → 订阅事件 → 收到 `session_started`
|
||||||
|
2. ✅ 飞行模式 30 秒 → 关闭飞行模式 → 自动重新订阅 → 收到事件
|
||||||
|
3. ✅ 应用后台 5 分钟 → 恢复前台 → Keep-Alive 保持连接 → 收到事件
|
||||||
|
4. ✅ 长时间空闲（30 分钟）→ 创建会话 → Keep-Alive 仍然工作
|
||||||
|
|
||||||
|
## 参考资料
|
||||||
|
|
||||||
|
### 官方文档
|
||||||
|
- [gRPC Keepalive Guide](https://grpc.io/docs/guides/keepalive/)
|
||||||
|
- [Android gRPC Guide](https://developer.android.com/guide/topics/connectivity/grpc)
|
||||||
|
- [Performance Best Practices](https://learn.microsoft.com/en-us/aspnet/core/grpc/performance)
|
||||||
|
|
||||||
|
### 关键 Issues
|
||||||
|
- [How to restart bi-directional stream after network disconnection](https://github.com/grpc/grpc-java/issues/8177)
|
||||||
|
- [Network connectivity changes on Android](https://github.com/grpc/grpc-java/issues/4011)
|
||||||
|
|
||||||
|
### 最新文章 (2026)
|
||||||
|
- [How to Implement gRPC Keepalive for Long-Lived Connections](https://oneuptime.com/blog/post/2026-01-08-grpc-keepalive-connections/view)
|
||||||
|
|
||||||
|
## 总结
|
||||||
|
|
||||||
|
### 当前问题根源
|
||||||
|
1. **设计错误**: 尝试"恢复"已关闭的流，但 gRPC 流无法恢复
|
||||||
|
2. **缺少 Keep-Alive**: 空闲连接被中间设备清理
|
||||||
|
3. **没有自动重启**: 流失败后需要手动重新发起
|
||||||
|
|
||||||
|
### 正确解决方案
|
||||||
|
1. ✅ 添加 Keep-Alive 配置（20s PING，5s 超时）
|
||||||
|
2. ✅ 保存流配置，失败后重新发起 RPC（不是"恢复"）
|
||||||
|
3. ✅ 使用 Flow.retryWhen 自动重启流
|
||||||
|
4. ✅ 监听网络状态，立即 resetConnectBackoff()
|
||||||
|
|
||||||
|
### 关键理念转变
|
||||||
|
```
|
||||||
|
旧思维: 连接 → 订阅流 → 断开 → 重连 → "恢复"流 ❌
|
||||||
|
新思维: 连接 → 订阅流 → 断开 → 重连 → "重新发起"流 ✅
|
||||||
|
```
|
||||||
|
|
||||||
|
**Flow 不是持久化对象，是一次性的数据流。断开后必须重新创建。**
|
||||||
|
|
@ -0,0 +1,251 @@
|
||||||
|
# Reconnection Event Stream Bug Analysis
|
||||||
|
|
||||||
|
## Problem Summary
|
||||||
|
|
||||||
|
After network disconnection and reconnection, the event stream subscription is NOT restored, causing:
|
||||||
|
- No `session_started` event received
|
||||||
|
- Keygen never starts
|
||||||
|
- Messages pile up forever (539 pending)
|
||||||
|
|
||||||
|
## Root Cause
|
||||||
|
|
||||||
|
### The Bug Chain
|
||||||
|
|
||||||
|
```
|
||||||
|
1. Network disconnects
|
||||||
|
↓
|
||||||
|
2. subscribeSessionEvents Flow closes
|
||||||
|
↓
|
||||||
|
3. awaitClose block executes (GrpcClient.kt:925)
|
||||||
|
eventStreamSubscribed.set(false) ← FLAG CLEARED
|
||||||
|
↓
|
||||||
|
4. Network reconnects successfully
|
||||||
|
↓
|
||||||
|
5. reSubscribeStreams() called (GrpcClient.kt:202)
|
||||||
|
↓
|
||||||
|
6. Line 506 checks:
|
||||||
|
val needsResubscribe = eventStreamSubscribed.get() || activeMessageSubscription != null
|
||||||
|
↓
|
||||||
|
7. eventStreamSubscribed.get() returns FALSE ❌
|
||||||
|
activeMessageSubscription is also NULL ❌
|
||||||
|
↓
|
||||||
|
8. needsResubscribe = false
|
||||||
|
↓
|
||||||
|
9. Callback NEVER invoked
|
||||||
|
↓
|
||||||
|
10. Event stream NEVER restored
|
||||||
|
```
|
||||||
|
|
||||||
|
## Code Evidence
|
||||||
|
|
||||||
|
### GrpcClient.kt - Where Flag is Set/Cleared
|
||||||
|
|
||||||
|
**Line 844** - Flag set when subscribing:
|
||||||
|
```kotlin
|
||||||
|
fun subscribeSessionEvents(partyId: String): Flow<SessionEventData> = callbackFlow {
|
||||||
|
eventStreamSubscribed.set(true) ← Set to TRUE
|
||||||
|
eventStreamPartyId = partyId
|
||||||
|
...
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Line 925** - Flag cleared when Flow closes (THE BUG):
|
||||||
|
```kotlin
|
||||||
|
awaitClose {
|
||||||
|
Log.d(TAG, "subscribeSessionEvents: Flow closed for partyId=$partyId")
|
||||||
|
eventStreamSubscribed.set(false) ← Set to FALSE on disconnect
|
||||||
|
eventStreamPartyId = null
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Line 506** - Reconnection check (FAILS because flag is false):
|
||||||
|
```kotlin
|
||||||
|
private fun reSubscribeStreams() {
|
||||||
|
val needsResubscribe = eventStreamSubscribed.get() || activeMessageSubscription != null
|
||||||
|
// ↑ Returns FALSE after disconnect
|
||||||
|
|
||||||
|
if (needsResubscribe) { ← This condition is FALSE
|
||||||
|
Log.d(TAG, "Triggering stream re-subscription callback")
|
||||||
|
...
|
||||||
|
onReconnectedCallback?.invoke() ← NEVER REACHED
|
||||||
|
}
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Log Evidence
|
||||||
|
|
||||||
|
### Normal Reconnection (16:28:26) - WORKS ✅
|
||||||
|
```
|
||||||
|
16:28:26.082 D/GrpcClient: Connected successfully
|
||||||
|
16:28:26.086 D/GrpcClient: Re-registering party: 7c72c28f...
|
||||||
|
16:28:26.130 D/GrpcClient: Party registered: 7c72c28f...
|
||||||
|
16:28:26.130 D/GrpcClient: Triggering stream re-subscription callback ← Present!
|
||||||
|
16:28:26.130 D/GrpcClient: - Event stream: true, partyId: 7c72c28f...
|
||||||
|
16:28:26.130 D/TssRepository: gRPC reconnected, restoring streams... ← Present!
|
||||||
|
16:28:26.130 D/TssRepository: Restoring session event subscription ← Present!
|
||||||
|
```
|
||||||
|
|
||||||
|
### Problem Reconnection (16:29:47) - FAILS ❌
|
||||||
|
```
|
||||||
|
16:29:47.090 D/GrpcClient: Connected successfully
|
||||||
|
16:29:47.093 D/GrpcClient: Re-registering party: 7c72c28f...
|
||||||
|
16:29:47.146 D/GrpcClient: Party registered: 7c72c28f...
|
||||||
|
[MISSING]: "Triggering stream re-subscription callback" ← NOT PRESENT!
|
||||||
|
[MISSING]: "gRPC reconnected, restoring streams..." ← NOT PRESENT!
|
||||||
|
[MISSING]: "Restoring session event subscription" ← NOT PRESENT!
|
||||||
|
|
||||||
|
Result:
|
||||||
|
16:30:47.198 W/GrpcClient: Has 539 pending messages - may have missed events
|
||||||
|
16:31:17.237 W/GrpcClient: Has 539 pending messages - may have missed events
|
||||||
|
```
|
||||||
|
|
||||||
|
## Why First Reconnection Worked
|
||||||
|
|
||||||
|
Looking at the timeline:
|
||||||
|
```
|
||||||
|
16:27:53 - App started, event subscription started
|
||||||
|
16:28:26 - First reconnect (1 minute later)
|
||||||
|
Event subscription was STILL ACTIVE
|
||||||
|
eventStreamSubscribed = true ✅
|
||||||
|
|
||||||
|
16:29:15 - Network disconnect (49 seconds later)
|
||||||
|
Flow closed → eventStreamSubscribed set to FALSE ❌
|
||||||
|
|
||||||
|
16:29:47 - Second reconnect
|
||||||
|
eventStreamSubscribed = false ❌
|
||||||
|
Callback NOT invoked ❌
|
||||||
|
```
|
||||||
|
|
||||||
|
**Key Insight**: The first reconnection worked because the event stream Flow hadn't closed yet. The second reconnection failed because the Flow had closed and cleared the flag.
|
||||||
|
|
||||||
|
## The Design Flaw
|
||||||
|
|
||||||
|
The current design has a **state tracking inconsistency**:
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
// When to subscribe?
|
||||||
|
eventStreamSubscribed = true // "I am currently subscribed"
|
||||||
|
|
||||||
|
// When to unsubscribe?
|
||||||
|
eventStreamSubscribed = false // "I am no longer subscribed"
|
||||||
|
|
||||||
|
// When to re-subscribe?
|
||||||
|
if (eventStreamSubscribed) { ... } // ❌ WRONG - flag is already false!
|
||||||
|
```
|
||||||
|
|
||||||
|
**Problem**: The flag tracks "am I currently subscribed?" but reconnection logic needs to know "should I re-subscribe?". These are two different concepts.
|
||||||
|
|
||||||
|
## Solution Options
|
||||||
|
|
||||||
|
### Option 1: Don't Clear Flag in awaitClose (Simple)
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
awaitClose {
|
||||||
|
Log.d(TAG, "subscribeSessionEvents: Flow closed for partyId=$partyId")
|
||||||
|
// DON'T clear the flag - keep it for reconnection
|
||||||
|
// eventStreamSubscribed.set(false) ← REMOVE THIS
|
||||||
|
// eventStreamPartyId = null ← REMOVE THIS
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Pros**: Minimal change, preserves intent to re-subscribe
|
||||||
|
**Cons**: Flag no longer accurately reflects current state
|
||||||
|
|
||||||
|
### Option 2: Add Separate "Should Restore" Flag (Better)
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
// Two separate flags:
|
||||||
|
private val eventStreamSubscribed = AtomicBoolean(false) // Current state
|
||||||
|
private val shouldRestoreEventStream = AtomicBoolean(false) // Intent to restore
|
||||||
|
|
||||||
|
// When subscribing:
|
||||||
|
eventStreamSubscribed.set(true)
|
||||||
|
shouldRestoreEventStream.set(true) // Remember to restore
|
||||||
|
|
||||||
|
// In awaitClose:
|
||||||
|
eventStreamSubscribed.set(false) // No longer subscribed
|
||||||
|
// Keep shouldRestoreEventStream = true // But should restore on reconnect
|
||||||
|
|
||||||
|
// In reSubscribeStreams:
|
||||||
|
val needsResubscribe = shouldRestoreEventStream.get() || activeMessageSubscription != null
|
||||||
|
```
|
||||||
|
|
||||||
|
**Pros**: Clear separation of concerns, accurate state tracking
|
||||||
|
**Cons**: More code, requires careful handling of clear conditions
|
||||||
|
|
||||||
|
### Option 3: Store Last Subscription State (Most Robust)
|
||||||
|
|
||||||
|
```kotlin
|
||||||
|
// Store full subscription state for recovery
|
||||||
|
private data class StreamState(
|
||||||
|
val eventStreamPartyId: String?,
|
||||||
|
val messageSessionId: String?,
|
||||||
|
val messagePartyId: String?
|
||||||
|
)
|
||||||
|
|
||||||
|
private val lastStreamState = AtomicReference<StreamState>(null)
|
||||||
|
|
||||||
|
// On subscribe, save state
|
||||||
|
// On reconnect, restore from saved state
|
||||||
|
```
|
||||||
|
|
||||||
|
**Pros**: Can restore exact previous state, handles complex scenarios
|
||||||
|
**Cons**: Most complex implementation
|
||||||
|
|
||||||
|
## Recommended Fix
|
||||||
|
|
||||||
|
**Use Option 1 (simplest) with Option 2 concept (clearer intent)**:
|
||||||
|
|
||||||
|
1. Don't clear `eventStreamSubscribed` in `awaitClose`
|
||||||
|
2. Only clear it when user explicitly unsubscribes or app shuts down
|
||||||
|
3. This preserves the "I was subscribed, so re-subscribe on reconnect" behavior
|
||||||
|
|
||||||
|
**Alternative**: Add explicit unsubscribe call only when intentionally stopping (not on disconnect).
|
||||||
|
|
||||||
|
## Files to Modify
|
||||||
|
|
||||||
|
### GrpcClient.kt
|
||||||
|
|
||||||
|
**Line 923-927** - Remove flag clearing in awaitClose:
|
||||||
|
```kotlin
|
||||||
|
awaitClose {
|
||||||
|
Log.d(TAG, "subscribeSessionEvents: Flow closed for partyId=$partyId")
|
||||||
|
// Keep flags for reconnection - don't clear here
|
||||||
|
// Only clear on explicit unsubscribe or disconnect
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
**Line 933-936** - Keep explicit unsubscribe as-is:
|
||||||
|
```kotlin
|
||||||
|
fun unsubscribeSessionEvents() {
|
||||||
|
eventStreamSubscribed.set(false)
|
||||||
|
eventStreamPartyId = null
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
## Testing Checklist
|
||||||
|
|
||||||
|
After fix:
|
||||||
|
- [ ] Start app, subscribe to events
|
||||||
|
- [ ] Simulate network disconnect (airplane mode)
|
||||||
|
- [ ] Verify log shows: "Triggering stream re-subscription callback"
|
||||||
|
- [ ] Verify log shows: "gRPC reconnected, restoring streams..."
|
||||||
|
- [ ] Verify log shows: "Restoring session event subscription"
|
||||||
|
- [ ] Verify pending messages start decreasing
|
||||||
|
- [ ] Test 2-of-3 keygen succeeds after reconnection
|
||||||
|
|
||||||
|
## Why This Wasn't Caught Before
|
||||||
|
|
||||||
|
1. **Timing-dependent**: Only fails if Flow closes before reconnect
|
||||||
|
2. **Works in most cases**: Quick reconnects (< 1 minute) often succeed before Flow timeout
|
||||||
|
3. **No explicit test**: Didn't test scenario of "disconnect → wait for Flow to close → reconnect"
|
||||||
|
4. **Silent failure**: No error logged, just missing callback invocation
|
||||||
|
|
||||||
|
## Conclusion
|
||||||
|
|
||||||
|
The safeLaunch optimization did NOT cause this bug. The bug exists because:
|
||||||
|
1. `awaitClose` clears `eventStreamSubscribed` on disconnect
|
||||||
|
2. Reconnection logic relies on this flag to decide if callback should be invoked
|
||||||
|
3. After disconnect, flag is false, so callback is never invoked
|
||||||
|
|
||||||
|
**Fix**: Don't clear the subscription intent flag on temporary disconnection.
|
||||||
|
|
@ -65,6 +65,13 @@ fun TssPartyApp(
|
||||||
viewModel: MainViewModel = hiltViewModel(),
|
viewModel: MainViewModel = hiltViewModel(),
|
||||||
onCopyToClipboard: (String) -> Unit = {}
|
onCopyToClipboard: (String) -> Unit = {}
|
||||||
) {
|
) {
|
||||||
|
val context = LocalContext.current
|
||||||
|
|
||||||
|
// Setup network monitoring once during initialization
|
||||||
|
LaunchedEffect(Unit) {
|
||||||
|
viewModel.setupNetworkMonitoring(context)
|
||||||
|
}
|
||||||
|
|
||||||
val navController = rememberNavController()
|
val navController = rememberNavController()
|
||||||
val appState by viewModel.appState.collectAsState()
|
val appState by viewModel.appState.collectAsState()
|
||||||
val uiState by viewModel.uiState.collectAsState()
|
val uiState by viewModel.uiState.collectAsState()
|
||||||
|
|
@ -119,7 +126,7 @@ fun TssPartyApp(
|
||||||
var transferWalletId by remember { mutableStateOf<Long?>(null) }
|
var transferWalletId by remember { mutableStateOf<Long?>(null) }
|
||||||
|
|
||||||
// Export/Import file handling
|
// Export/Import file handling
|
||||||
val context = LocalContext.current
|
// Note: context is already declared at the top of the function
|
||||||
// Use rememberSaveable to persist across configuration changes (e.g., file picker activity)
|
// Use rememberSaveable to persist across configuration changes (e.g., file picker activity)
|
||||||
var pendingExportJson by rememberSaveable { mutableStateOf<String?>(null) }
|
var pendingExportJson by rememberSaveable { mutableStateOf<String?>(null) }
|
||||||
var pendingExportAddress by rememberSaveable { mutableStateOf<String?>(null) }
|
var pendingExportAddress by rememberSaveable { mutableStateOf<String?>(null) }
|
||||||
|
|
|
||||||
|
|
@ -1,5 +1,10 @@
|
||||||
package com.durian.tssparty.data.remote
|
package com.durian.tssparty.data.remote
|
||||||
|
|
||||||
|
import android.content.Context
|
||||||
|
import android.net.ConnectivityManager
|
||||||
|
import android.net.Network
|
||||||
|
import android.net.NetworkCapabilities
|
||||||
|
import android.net.NetworkRequest
|
||||||
import android.util.Base64
|
import android.util.Base64
|
||||||
import android.util.Log
|
import android.util.Log
|
||||||
import com.durian.tssparty.domain.model.Participant
|
import com.durian.tssparty.domain.model.Participant
|
||||||
|
|
@ -101,24 +106,100 @@ class GrpcClient @Inject constructor() {
|
||||||
private var heartbeatJob: Job? = null
|
private var heartbeatJob: Job? = null
|
||||||
private val heartbeatFailCount = AtomicInteger(0)
|
private val heartbeatFailCount = AtomicInteger(0)
|
||||||
|
|
||||||
// Stream subscriptions (for recovery after reconnect)
|
|
||||||
private var activeMessageSubscription: MessageSubscription? = null
|
|
||||||
private var eventStreamSubscribed = AtomicBoolean(false)
|
|
||||||
private var eventStreamPartyId: String? = null
|
|
||||||
|
|
||||||
// Stream version tracking (to detect stale stream events)
|
// Stream version tracking (to detect stale stream events)
|
||||||
private val messageStreamVersion = AtomicInteger(0)
|
private val messageStreamVersion = AtomicInteger(0)
|
||||||
private val eventStreamVersion = AtomicInteger(0)
|
private val eventStreamVersion = AtomicInteger(0)
|
||||||
|
|
||||||
// Reconnection callback - called when streams need to be re-established
|
|
||||||
private var onReconnectedCallback: (() -> Unit)? = null
|
|
||||||
|
|
||||||
// Channel state monitoring
|
// Channel state monitoring
|
||||||
private var channelStateMonitorJob: Job? = null
|
private var channelStateMonitorJob: Job? = null
|
||||||
|
|
||||||
// Coroutine scope for background tasks
|
// Coroutine scope for background tasks
|
||||||
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
private val scope = CoroutineScope(Dispatchers.IO + SupervisorJob())
|
||||||
|
|
||||||
|
// Network monitoring
|
||||||
|
private var networkCallback: ConnectivityManager.NetworkCallback? = null
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Setup network state monitoring to accelerate reconnection
|
||||||
|
*
|
||||||
|
* Android network state monitoring is critical for reliable gRPC connections:
|
||||||
|
* 1. When network becomes available, call channel.resetConnectBackoff()
|
||||||
|
* 2. This bypasses the default 60-second DNS resolution backoff
|
||||||
|
* 3. Enables immediate reconnection instead of waiting
|
||||||
|
*
|
||||||
|
* Reference: https://github.com/grpc/grpc-java/issues/4011
|
||||||
|
*
|
||||||
|
* @param context Application or Activity context
|
||||||
|
*/
|
||||||
|
fun setupNetworkMonitoring(context: Context) {
|
||||||
|
// Unregister old callback if exists
|
||||||
|
networkCallback?.let { callback ->
|
||||||
|
try {
|
||||||
|
val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
|
||||||
|
connectivityManager?.unregisterNetworkCallback(callback)
|
||||||
|
} catch (e: Exception) {
|
||||||
|
Log.e(TAG, "Failed to unregister old network callback", e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
|
||||||
|
if (connectivityManager == null) {
|
||||||
|
Log.e(TAG, "ConnectivityManager not available")
|
||||||
|
return
|
||||||
|
}
|
||||||
|
|
||||||
|
val callback = object : ConnectivityManager.NetworkCallback() {
|
||||||
|
override fun onAvailable(network: Network) {
|
||||||
|
Log.d(TAG, "Network available, resetting connect backoff for immediate reconnection")
|
||||||
|
// CRITICAL: Reset backoff to avoid 60-second DNS resolution delay
|
||||||
|
channel?.resetConnectBackoff()
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun onLost(network: Network) {
|
||||||
|
Log.w(TAG, "Network lost")
|
||||||
|
}
|
||||||
|
|
||||||
|
override fun onCapabilitiesChanged(network: Network, networkCapabilities: NetworkCapabilities) {
|
||||||
|
val hasInternet = networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
|
||||||
|
val isValidated = networkCapabilities.hasCapability(NetworkCapabilities.NET_CAPABILITY_VALIDATED)
|
||||||
|
Log.d(TAG, "Network capabilities changed: hasInternet=$hasInternet, isValidated=$isValidated")
|
||||||
|
|
||||||
|
// Reset backoff when network becomes validated (has actual internet connectivity)
|
||||||
|
if (hasInternet && isValidated) {
|
||||||
|
channel?.resetConnectBackoff()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
val request = NetworkRequest.Builder()
|
||||||
|
.addCapability(NetworkCapabilities.NET_CAPABILITY_INTERNET)
|
||||||
|
.build()
|
||||||
|
|
||||||
|
try {
|
||||||
|
connectivityManager.registerNetworkCallback(request, callback)
|
||||||
|
networkCallback = callback
|
||||||
|
Log.d(TAG, "Network monitoring registered successfully")
|
||||||
|
} catch (e: Exception) {
|
||||||
|
Log.e(TAG, "Failed to register network callback", e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Stop network monitoring (call in cleanup)
|
||||||
|
*/
|
||||||
|
fun stopNetworkMonitoring(context: Context) {
|
||||||
|
networkCallback?.let { callback ->
|
||||||
|
try {
|
||||||
|
val connectivityManager = context.getSystemService(Context.CONNECTIVITY_SERVICE) as? ConnectivityManager
|
||||||
|
connectivityManager?.unregisterNetworkCallback(callback)
|
||||||
|
networkCallback = null
|
||||||
|
Log.d(TAG, "Network monitoring stopped")
|
||||||
|
} catch (e: Exception) {
|
||||||
|
Log.e(TAG, "Failed to unregister network callback", e)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Connect to the Message Router server
|
* Connect to the Message Router server
|
||||||
*/
|
*/
|
||||||
|
|
@ -142,10 +223,11 @@ class GrpcClient @Inject constructor() {
|
||||||
|
|
||||||
val builder = ManagedChannelBuilder
|
val builder = ManagedChannelBuilder
|
||||||
.forAddress(host, port)
|
.forAddress(host, port)
|
||||||
.keepAliveTime(30, TimeUnit.SECONDS)
|
// Keep-Alive configuration for stable long-lived connections
|
||||||
.keepAliveTimeout(10, TimeUnit.SECONDS)
|
.keepAliveTime(20, TimeUnit.SECONDS) // Send PING every 20 seconds
|
||||||
.keepAliveWithoutCalls(true)
|
.keepAliveTimeout(5, TimeUnit.SECONDS) // 5 seconds to wait for ACK
|
||||||
.idleTimeout(5, TimeUnit.MINUTES)
|
.keepAliveWithoutCalls(true) // Keep pinging even without active RPCs
|
||||||
|
.idleTimeout(Long.MAX_VALUE, TimeUnit.DAYS) // Never timeout idle connections
|
||||||
|
|
||||||
// Use TLS for port 443, plaintext for other ports (like local development)
|
// Use TLS for port 443, plaintext for other ports (like local development)
|
||||||
if (port == 443) {
|
if (port == 443) {
|
||||||
|
|
@ -198,8 +280,8 @@ class GrpcClient @Inject constructor() {
|
||||||
// Restart heartbeat
|
// Restart heartbeat
|
||||||
startHeartbeat()
|
startHeartbeat()
|
||||||
|
|
||||||
// Re-subscribe to streams
|
// Emit reconnected event for StreamManager to restart streams
|
||||||
reSubscribeStreams()
|
_connectionEvents.tryEmit(GrpcConnectionEvent.Reconnected)
|
||||||
|
|
||||||
return@withTimeout
|
return@withTimeout
|
||||||
}
|
}
|
||||||
|
|
@ -498,65 +580,6 @@ class GrpcClient @Inject constructor() {
|
||||||
return false
|
return false
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Re-subscribe to streams after reconnection
|
|
||||||
* Notifies the repository layer to re-establish message/event subscriptions
|
|
||||||
*/
|
|
||||||
private fun reSubscribeStreams() {
|
|
||||||
val needsResubscribe = eventStreamSubscribed.get() || activeMessageSubscription != null
|
|
||||||
|
|
||||||
if (needsResubscribe) {
|
|
||||||
Log.d(TAG, "Triggering stream re-subscription callback")
|
|
||||||
Log.d(TAG, " - Event stream: ${eventStreamSubscribed.get()}, partyId: $eventStreamPartyId")
|
|
||||||
Log.d(TAG, " - Message stream: ${activeMessageSubscription?.sessionId}")
|
|
||||||
|
|
||||||
// Notify repository to re-establish streams
|
|
||||||
scope.launch {
|
|
||||||
// Wait for channel to be fully ready instead of fixed delay
|
|
||||||
if (waitForChannelReady()) {
|
|
||||||
try {
|
|
||||||
onReconnectedCallback?.invoke()
|
|
||||||
// Emit reconnected event
|
|
||||||
_connectionEvents.tryEmit(GrpcConnectionEvent.Reconnected)
|
|
||||||
} catch (e: Exception) {
|
|
||||||
Log.e(TAG, "Reconnect callback failed: ${e.message}")
|
|
||||||
// Don't let callback failure affect the connection state
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
Log.w(TAG, "Channel not ready for stream re-subscription, skipping")
|
|
||||||
}
|
|
||||||
}
|
|
||||||
} else {
|
|
||||||
// No streams to restore, still emit reconnected event
|
|
||||||
_connectionEvents.tryEmit(GrpcConnectionEvent.Reconnected)
|
|
||||||
}
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Set callback for reconnection events
|
|
||||||
* Called when connection is restored and streams need to be re-established
|
|
||||||
*/
|
|
||||||
fun setOnReconnectedCallback(callback: () -> Unit) {
|
|
||||||
onReconnectedCallback = callback
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Get active message subscription info (for recovery)
|
|
||||||
*/
|
|
||||||
fun getActiveMessageSubscription(): Pair<String, String>? {
|
|
||||||
return activeMessageSubscription?.let { Pair(it.sessionId, it.partyId) }
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Check if event stream was subscribed (for recovery)
|
|
||||||
*/
|
|
||||||
fun wasEventStreamSubscribed(): Boolean = eventStreamSubscribed.get()
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Get event stream party ID (for recovery)
|
|
||||||
*/
|
|
||||||
fun getEventStreamPartyId(): String? = eventStreamPartyId
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Check if connected
|
* Check if connected
|
||||||
*/
|
*/
|
||||||
|
|
@ -764,9 +787,6 @@ class GrpcClient @Inject constructor() {
|
||||||
* Subscribe to messages for a party (with auto-recovery)
|
* Subscribe to messages for a party (with auto-recovery)
|
||||||
*/
|
*/
|
||||||
fun subscribeMessages(sessionId: String, partyId: String): Flow<IncomingMessage> = callbackFlow {
|
fun subscribeMessages(sessionId: String, partyId: String): Flow<IncomingMessage> = callbackFlow {
|
||||||
// Save subscription for recovery
|
|
||||||
activeMessageSubscription = MessageSubscription(sessionId, partyId)
|
|
||||||
|
|
||||||
// Capture current stream version to detect stale callbacks
|
// Capture current stream version to detect stale callbacks
|
||||||
val streamVersion = messageStreamVersion.incrementAndGet()
|
val streamVersion = messageStreamVersion.incrementAndGet()
|
||||||
|
|
||||||
|
|
@ -820,11 +840,8 @@ class GrpcClient @Inject constructor() {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stream ended unexpectedly - trigger reconnect if we should still be subscribed
|
// Stream ended - StreamManager will handle reconnection
|
||||||
if (activeMessageSubscription != null && shouldReconnect.get()) {
|
Log.d(TAG, "Message stream ended")
|
||||||
Log.w(TAG, "Message stream ended unexpectedly, triggering reconnect")
|
|
||||||
triggerReconnect("Message stream ended")
|
|
||||||
}
|
|
||||||
close()
|
close()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -832,7 +849,7 @@ class GrpcClient @Inject constructor() {
|
||||||
asyncStub?.subscribeMessages(request, observer)
|
asyncStub?.subscribeMessages(request, observer)
|
||||||
|
|
||||||
awaitClose {
|
awaitClose {
|
||||||
activeMessageSubscription = null
|
Log.d(TAG, "subscribeMessages: Flow closed for sessionId=$sessionId")
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -840,10 +857,6 @@ class GrpcClient @Inject constructor() {
|
||||||
* Subscribe to session events (with auto-recovery)
|
* Subscribe to session events (with auto-recovery)
|
||||||
*/
|
*/
|
||||||
fun subscribeSessionEvents(partyId: String): Flow<SessionEventData> = callbackFlow {
|
fun subscribeSessionEvents(partyId: String): Flow<SessionEventData> = callbackFlow {
|
||||||
// Save subscription for recovery
|
|
||||||
eventStreamSubscribed.set(true)
|
|
||||||
eventStreamPartyId = partyId
|
|
||||||
|
|
||||||
// Capture current stream version to detect stale callbacks
|
// Capture current stream version to detect stale callbacks
|
||||||
val streamVersion = eventStreamVersion.incrementAndGet()
|
val streamVersion = eventStreamVersion.incrementAndGet()
|
||||||
|
|
||||||
|
|
@ -900,11 +913,8 @@ class GrpcClient @Inject constructor() {
|
||||||
return
|
return
|
||||||
}
|
}
|
||||||
|
|
||||||
// Stream ended unexpectedly - trigger reconnect if we should still be subscribed
|
// Stream ended - StreamManager will handle reconnection
|
||||||
if (eventStreamSubscribed.get() && shouldReconnect.get()) {
|
Log.d(TAG, "Event stream ended")
|
||||||
Log.w(TAG, "Event stream ended unexpectedly, triggering reconnect")
|
|
||||||
triggerReconnect("Event stream ended")
|
|
||||||
}
|
|
||||||
close()
|
close()
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
@ -922,26 +932,9 @@ class GrpcClient @Inject constructor() {
|
||||||
|
|
||||||
awaitClose {
|
awaitClose {
|
||||||
Log.d(TAG, "subscribeSessionEvents: Flow closed for partyId=$partyId")
|
Log.d(TAG, "subscribeSessionEvents: Flow closed for partyId=$partyId")
|
||||||
eventStreamSubscribed.set(false)
|
|
||||||
eventStreamPartyId = null
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Unsubscribe from session events
|
|
||||||
*/
|
|
||||||
fun unsubscribeSessionEvents() {
|
|
||||||
eventStreamSubscribed.set(false)
|
|
||||||
eventStreamPartyId = null
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
|
||||||
* Unsubscribe from messages
|
|
||||||
*/
|
|
||||||
fun unsubscribeMessages() {
|
|
||||||
activeMessageSubscription = null
|
|
||||||
}
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Report completion
|
* Report completion
|
||||||
*/
|
*/
|
||||||
|
|
@ -1009,13 +1002,6 @@ class GrpcClient @Inject constructor() {
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
|
||||||
* Message subscription info
|
|
||||||
*/
|
|
||||||
private data class MessageSubscription(
|
|
||||||
val sessionId: String,
|
|
||||||
val partyId: String
|
|
||||||
)
|
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Data class for join session response
|
* Data class for join session response
|
||||||
|
|
|
||||||
|
|
@ -0,0 +1,282 @@
|
||||||
|
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,6 +14,7 @@ import com.durian.tssparty.data.remote.GrpcConnectionState
|
||||||
import com.durian.tssparty.data.remote.IncomingMessage
|
import com.durian.tssparty.data.remote.IncomingMessage
|
||||||
import com.durian.tssparty.data.remote.JoinSessionData
|
import com.durian.tssparty.data.remote.JoinSessionData
|
||||||
import com.durian.tssparty.data.remote.SessionEventData
|
import com.durian.tssparty.data.remote.SessionEventData
|
||||||
|
import com.durian.tssparty.data.remote.StreamManager
|
||||||
import com.durian.tssparty.domain.model.*
|
import com.durian.tssparty.domain.model.*
|
||||||
import com.durian.tssparty.util.AddressUtils
|
import com.durian.tssparty.util.AddressUtils
|
||||||
import com.durian.tssparty.util.TransactionUtils
|
import com.durian.tssparty.util.TransactionUtils
|
||||||
|
|
@ -213,6 +214,33 @@ class TssRepository @Inject constructor(
|
||||||
coroutineExceptionHandler
|
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 {
|
companion object {
|
||||||
// Job 名称常量
|
// Job 名称常量
|
||||||
private const val JOB_MESSAGE_COLLECTION = "message_collection"
|
private const val JOB_MESSAGE_COLLECTION = "message_collection"
|
||||||
|
|
@ -289,32 +317,16 @@ class TssRepository @Inject constructor(
|
||||||
private var accountServiceUrl: String = "https://rwaapi.szaiai.com"
|
private var accountServiceUrl: String = "https://rwaapi.szaiai.com"
|
||||||
|
|
||||||
init {
|
init {
|
||||||
// Set up reconnection callback to restore streams
|
// Monitor gRPC reconnection events and restart streams
|
||||||
grpcClient.setOnReconnectedCallback {
|
// IMPORTANT: Don't use callback pattern - use event Flow for better reliability
|
||||||
android.util.Log.d("TssRepository", "gRPC reconnected, restoring streams...")
|
repositoryScope.launch {
|
||||||
restoreStreamsAfterReconnect()
|
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)
|
||||||
* Restore message and event streams after gRPC reconnection
|
streamManager.restartAllStreams()
|
||||||
*/
|
}
|
||||||
private fun restoreStreamsAfterReconnect() {
|
|
||||||
val sessionId = currentMessageRoutingSessionId
|
|
||||||
val partyIndex = currentMessageRoutingPartyIndex
|
|
||||||
val routingPartyId = currentMessageRoutingPartyId
|
|
||||||
|
|
||||||
// Restore message routing if we had an active session
|
|
||||||
if (sessionId != null && partyIndex != null) {
|
|
||||||
android.util.Log.d("TssRepository", "Restoring message routing for session: $sessionId, routingPartyId: $routingPartyId")
|
|
||||||
startMessageRouting(sessionId, partyIndex, routingPartyId)
|
|
||||||
}
|
|
||||||
|
|
||||||
// Restore session event subscription with the correct partyId
|
|
||||||
if (grpcClient.wasEventStreamSubscribed()) {
|
|
||||||
val eventPartyId = currentSessionEventPartyId
|
|
||||||
android.util.Log.d("TssRepository", "Restoring session event subscription with partyId: $eventPartyId")
|
|
||||||
startSessionEventSubscription(eventPartyId)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -362,6 +374,18 @@ class TssRepository @Inject constructor(
|
||||||
grpcClient.connect(host, port)
|
grpcClient.connect(host, port)
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Setup network state monitoring for reliable reconnection
|
||||||
|
*
|
||||||
|
* This should be called once during app initialization (e.g., in MainActivity.onCreate)
|
||||||
|
* to enable immediate reconnection when network becomes available.
|
||||||
|
*
|
||||||
|
* @param context Application or Activity context
|
||||||
|
*/
|
||||||
|
fun setupNetworkMonitoring(context: android.content.Context) {
|
||||||
|
grpcClient.setupNetworkMonitoring(context)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Disconnect from the server
|
* Disconnect from the server
|
||||||
*/
|
*/
|
||||||
|
|
@ -483,9 +507,10 @@ class TssRepository @Inject constructor(
|
||||||
currentSessionEventPartyId = effectivePartyId
|
currentSessionEventPartyId = effectivePartyId
|
||||||
android.util.Log.d("TssRepository", "Starting session event subscription for partyId: $effectivePartyId (device partyId: $devicePartyId)")
|
android.util.Log.d("TssRepository", "Starting session event subscription for partyId: $effectivePartyId (device partyId: $devicePartyId)")
|
||||||
|
|
||||||
// 使用 JobManager 启动（自动取消同名旧 Job）
|
// Use StreamManager for reliable stream management with auto-reconnection
|
||||||
jobManager.launch(JOB_SESSION_EVENT) {
|
streamManager.startEventStream(
|
||||||
grpcClient.subscribeSessionEvents(effectivePartyId).collect { event ->
|
partyId = effectivePartyId,
|
||||||
|
onEvent = { event ->
|
||||||
android.util.Log.d("TssRepository", "=== Session event received ===")
|
android.util.Log.d("TssRepository", "=== Session event received ===")
|
||||||
android.util.Log.d("TssRepository", " eventType: ${event.eventType}")
|
android.util.Log.d("TssRepository", " eventType: ${event.eventType}")
|
||||||
android.util.Log.d("TssRepository", " sessionId: ${event.sessionId}")
|
android.util.Log.d("TssRepository", " sessionId: ${event.sessionId}")
|
||||||
|
|
@ -570,8 +595,12 @@ class TssRepository @Inject constructor(
|
||||||
android.util.Log.d("TssRepository", " Reason: sessionId mismatch (event: ${event.sessionId}, active: ${activeSession.sessionId})")
|
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
|
||||||
}
|
}
|
||||||
}
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
@ -585,8 +614,8 @@ class TssRepository @Inject constructor(
|
||||||
* partyId from keygen (shareEntity.partyId).
|
* partyId from keygen (shareEntity.partyId).
|
||||||
*/
|
*/
|
||||||
private fun ensureSessionEventSubscriptionActive(signingPartyId: String? = null) {
|
private fun ensureSessionEventSubscriptionActive(signingPartyId: String? = null) {
|
||||||
// Check if the session event job is still active using JobManager
|
// Check if the event stream is still active using StreamManager
|
||||||
val isActive = jobManager.isActive(JOB_SESSION_EVENT)
|
val isActive = streamManager.isEventStreamActive()
|
||||||
val devicePartyId = requirePartyId() // Ensure partyId is initialized
|
val devicePartyId = requirePartyId() // Ensure partyId is initialized
|
||||||
val effectivePartyId = signingPartyId ?: currentSessionEventPartyId ?: devicePartyId
|
val effectivePartyId = signingPartyId ?: currentSessionEventPartyId ?: devicePartyId
|
||||||
android.util.Log.d("TssRepository", "Checking session event subscription: isActive=$isActive, effectivePartyId=$effectivePartyId")
|
android.util.Log.d("TssRepository", "Checking session event subscription: isActive=$isActive, effectivePartyId=$effectivePartyId")
|
||||||
|
|
@ -595,16 +624,14 @@ class TssRepository @Inject constructor(
|
||||||
android.util.Log.w("TssRepository", "Session event subscription is not active, restarting...")
|
android.util.Log.w("TssRepository", "Session event subscription is not active, restarting...")
|
||||||
startSessionEventSubscription(signingPartyId)
|
startSessionEventSubscription(signingPartyId)
|
||||||
} else {
|
} else {
|
||||||
// Even if the job is "active", the gRPC stream may have silently disconnected
|
// Check if we need to switch to a different partyId for signing
|
||||||
// Force a restart to ensure we have a fresh connection
|
|
||||||
// Also restart if we need to switch to a different partyId for signing
|
|
||||||
val needsRestart = signingPartyId != null && signingPartyId != currentSessionEventPartyId
|
val needsRestart = signingPartyId != null && signingPartyId != currentSessionEventPartyId
|
||||||
if (needsRestart) {
|
if (needsRestart) {
|
||||||
android.util.Log.d("TssRepository", "Switching session event subscription to signingPartyId: $signingPartyId")
|
android.util.Log.d("TssRepository", "Switching session event subscription to signingPartyId: $signingPartyId")
|
||||||
|
startSessionEventSubscription(signingPartyId)
|
||||||
} else {
|
} else {
|
||||||
android.util.Log.d("TssRepository", "Refreshing session event subscription to ensure fresh connection")
|
android.util.Log.d("TssRepository", "Event stream is active with correct partyId, no restart needed")
|
||||||
}
|
}
|
||||||
startSessionEventSubscription(signingPartyId)
|
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
|
@ -2038,40 +2065,50 @@ class TssRepository @Inject constructor(
|
||||||
// Save for reconnection recovery
|
// Save for reconnection recovery
|
||||||
currentMessageRoutingPartyId = effectivePartyId
|
currentMessageRoutingPartyId = effectivePartyId
|
||||||
|
|
||||||
|
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_COLLECTION) {
|
||||||
android.util.Log.d("TssRepository", "Starting message routing: sessionId=$sessionId, routingPartyId=$effectivePartyId")
|
tssNativeBridge.outgoingMessages.collect { message ->
|
||||||
|
val payload = Base64.decode(message.payload, Base64.NO_WRAP)
|
||||||
// Collect outgoing messages from TSS and route via gRPC
|
grpcClient.routeMessage(
|
||||||
launch {
|
sessionId = sessionId,
|
||||||
tssNativeBridge.outgoingMessages.collect { message ->
|
fromParty = effectivePartyId, // Use the correct partyId for routing
|
||||||
val payload = Base64.decode(message.payload, Base64.NO_WRAP)
|
toParties = message.toParties ?: emptyList(),
|
||||||
grpcClient.routeMessage(
|
roundNumber = 0,
|
||||||
sessionId = sessionId,
|
messageType = if (message.isBroadcast) "broadcast" else "p2p",
|
||||||
fromParty = effectivePartyId, // Use the correct partyId for routing
|
payload = payload
|
||||||
toParties = message.toParties ?: emptyList(),
|
)
|
||||||
roundNumber = 0,
|
|
||||||
messageType = if (message.isBroadcast) "broadcast" else "p2p",
|
|
||||||
payload = payload
|
|
||||||
)
|
|
||||||
}
|
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
// Collect incoming messages from gRPC and send to TSS
|
// Part 2: Subscribe to incoming messages from gRPC and send to TSS
|
||||||
launch {
|
// Use StreamManager for reliable gRPC stream management with auto-reconnection
|
||||||
grpcClient.subscribeMessages(sessionId, effectivePartyId).collect { message ->
|
streamManager.startMessageStream(
|
||||||
// Find party index from party ID
|
sessionId = sessionId,
|
||||||
val session = _currentSession.value
|
partyId = effectivePartyId,
|
||||||
val fromPartyIndex = session?.participants?.find { it.partyId == message.fromParty }?.partyIndex
|
partyIndex = partyIndex,
|
||||||
?: return@collect
|
onMessage = { message ->
|
||||||
|
// Find party index from party ID
|
||||||
|
val session = _currentSession.value
|
||||||
|
val fromPartyIndex = session?.participants?.find { it.partyId == message.fromParty }?.partyIndex
|
||||||
|
|
||||||
|
if (fromPartyIndex != null) {
|
||||||
tssNativeBridge.sendIncomingMessage(
|
tssNativeBridge.sendIncomingMessage(
|
||||||
fromPartyIndex = fromPartyIndex,
|
fromPartyIndex = fromPartyIndex,
|
||||||
isBroadcast = message.isBroadcast,
|
isBroadcast = message.isBroadcast,
|
||||||
payload = message.payload
|
payload = message.payload
|
||||||
)
|
)
|
||||||
|
} 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
|
||||||
}
|
}
|
||||||
}
|
)
|
||||||
}
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
|
|
|
||||||
|
|
@ -231,6 +231,18 @@ class MainViewModel @Inject constructor(
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
/**
|
||||||
|
* Setup network state monitoring for reliable reconnection
|
||||||
|
*
|
||||||
|
* This enables immediate reconnection when network becomes available
|
||||||
|
* instead of waiting for DNS resolution backoff (up to 60 seconds).
|
||||||
|
*
|
||||||
|
* Should be called once during app initialization.
|
||||||
|
*/
|
||||||
|
fun setupNetworkMonitoring(context: android.content.Context) {
|
||||||
|
repository.setupNetworkMonitoring(context)
|
||||||
|
}
|
||||||
|
|
||||||
/**
|
/**
|
||||||
* Connect to Message Router server
|
* Connect to Message Router server
|
||||||
*/
|
*/
|
||||||
|
|
|
||||||
Loading…
Reference in New Issue