fix(reconnect): 上一次会话的关闭码污染下一次会话（v0.2.1）+ 重连链路的真实 WebSocket 集成测试
ReconnectPolicyTest（11 项）验的是**纯函数**「给定关闭码应该怎么退避」，而 ⑦
「连接频繁重连」的缺陷往往出在**调用点**。新增 PoseWebSocketReconnectTest（8 项）：
用 MockWebServer.withWebSocketUpgrade 在 JVM 上起**真实** WebSocket 服务端
（能按 §3.2 回 welcome、按 §3.6 回 pong、按指定关闭码关连接），无需设备即可覆盖
「该重连的」与「不该重连的」两条分支。Robolectric 只用来提供 android.os.Build 桩。
它抓到的真实缺陷（本版修复）：
上一次会话的关闭码会污染下一次会话，导致重新开始推流后失去自动重连能力。
复现：第一次连接被对端以 1001 关闭（按 §7 不重连，正确）→ 用户再点「开始推流」
→ 这一次是网络层失败（服务端还没起来），按 §7 **应当**重连 → 但 connect() 只重置了
reconnectCounter 与 lastPongAtMs，**没有清 lastCloseCode** → scheduleReconnect()
仍按 1001 判定 → 直接 stopped，**再也不会自动重连**。
测试失败信息（修复前）：
✗ 上一次 1001 关闭后重新连接，遇到网络失败仍应按策略重连
AssertionError: 当前 state=STOPPED，lastCloseCode=1001，attempts=0
修复（PoseWebSocketClient.kt）：
* connect()：清掉会话级关闭状态（lastCloseCode=0、lastCloseReason=""）；
* onFailure()：记录 recordClose(0, …)（契约 §7 中 0=未收到关闭帧 → 应重连），
顺带让界面「最近一次断开原因」显示本次失败而不是旧值。
用例清单：握手→streaming 与 welcome 保活参数解析、按协商间隔发 ping 收 pong、
1011 按退避重连且等待落在策略 ±20% 区间、成功后计数清零、1001/1002 不重连、
主动断开不重连、上一次 1001 后再连遇网络失败仍应重连（本次回归锁）、
服务端未启动时 1.5s 内 ≤3 次尝试（退避生效）。
实测：./gradlew :app:testDebugUnitTest → 6 个测试类 / 64 项用例 / 0 失败
（原 56 项）；build/verify.sh 全绿（签名、APK 版本元数据 0.2.1/201、native 库体检、
ABI、R8、dex）。版本号 0.2.0/200 → 0.2.1/201。
踩坑：第一版 MockWebServer 夹具的 101 响应没有回显 Sec-WebSocket-Protocol
（RFC 6455 要求），客户端记录的 negotiatedProtocol 为空导致断言失败 —— 那是
**夹具**不合规而非客户端问题，已在夹具里补上。写「真实对端」夹具时要按 RFC/契约
把服务端做对，否则会把夹具缺陷误判成产品缺陷。
This commit is contained in:
@@ -9,20 +9,41 @@ OpenDance Studio **阶段三**: Android 辅助 App(Kotlin + CameraX + MediaPipe
|
||||
|
||||
## 状态
|
||||
|
||||
**阶段三已完成**: T-AND-001 -> T-AND-013 全部落地（含重连状态机、MediaPipe 崩溃修复、APK 产物版本元数据校验）。
|
||||
**阶段三已完成**: T-AND-001 -> T-AND-016 全部落地（含重连状态机与**真实 WebSocket 重连链路测试**、MediaPipe 崩溃修复、APK 产物版本元数据校验、native 库体检）。
|
||||
|
||||
| 验收项 | 结果 |
|
||||
|--|--|
|
||||
| 单元测试 | OK **56 项全部通过**(JVM，无需设备；新增重连策略与 MediaPipe 诊断用例) |
|
||||
| 单元测试 | OK **64 项全部通过**(JVM，无需设备；含 MockWebServer 真实 WebSocket 的重连链路集成测试) |
|
||||
| **跨端金标准** | OK Kotlin 序列化器与 Web 端 TypeScript 实现**逐位一致** |
|
||||
| `./gradlew assembleRelease` | OK 生成**已签名** APK(28.6 MB) |
|
||||
| 签名校验 | OK `apksigner` 退出码 0, v2 方案，CN=OpenDance Studio Dev |
|
||||
| 包信息 | OK `com.opendance.studio` v0.2.0, minSdk 29(Android 10+)/ targetSdk 34 |
|
||||
| 包信息 | OK `com.opendance.studio` v0.2.1 (versionCode 201), minSdk 29(Android 10+)/ targetSdk 34 |
|
||||
| 模型随包分发 | OK `assets/pose_landmarker_lite.task` 5.7 MB(Stored 未压缩) |
|
||||
| ABI 覆盖 | OK arm64-v8a + armeabi-v7a(debug 额外含 x86/x86_64) |
|
||||
| APK 版本元数据 | OK 直接读产物 `AndroidManifest` 的 `versionName=0.2.0`/`versionCode=200`，与 `project.yaml` 一致（T-AND-013） |
|
||||
| APK 版本元数据 | OK 直接读产物 `AndroidManifest` 的 `versionName=0.2.1`/`versionCode=201`，与 `project.yaml` 一致（T-AND-013） |
|
||||
| native 库体检 | OK ELF 类型匹配双 ABI、`DT_NEEDED` 闭包完整（无缺失依赖）、`PT_LOAD` 页对齐 `0x4000`（16 KB 页兼容）、`zipalign -P 16` 通过（T-AND-015） |
|
||||
|
||||
## v0.2.1 修订（重连链路的真实 WebSocket 集成测试 + 一处真实缺陷）
|
||||
|
||||
**缺陷**：上一次会话的关闭码会**污染下一次会话**，导致重新开始推流后失去自动重连能力。
|
||||
复现路径：第一次连接被对端以 `1001` 关闭（按契约 §7 **不重连**，正确）→ 用户再次点
|
||||
「开始推流」→ 这一次是**网络层失败**（服务端还没起来），按 §7 **应当**重连 →
|
||||
但 `connect()` 只重置了重连计数与 pong 时间，**没有清 `lastCloseCode`**，
|
||||
`scheduleReconnect()` 仍按 1001 判定 → 直接 `stopped`，**再也不会自动重连**。
|
||||
|
||||
**修复**：`connect()` 清掉会话级关闭状态（`lastCloseCode = 0`、`lastCloseReason = ""`）；
|
||||
`onFailure()` 记录 `recordClose(0, …)`（0 = 未收到关闭帧 → 应重连），
|
||||
顺带让界面「最近一次断开原因」显示本次失败而不是旧值。
|
||||
|
||||
**为什么之前没发现**：`ReconnectPolicyTest`（11 项）验的是**纯函数**「给定关闭码怎么退避」，
|
||||
而缺陷在**调用点**（关闭码没被正确重置）。新增
|
||||
`app/src/test/java/com/opendance/studio/PoseWebSocketReconnectTest.kt`（8 项）：
|
||||
用 `MockWebServer.withWebSocketUpgrade` 在 JVM 上起**真实** WebSocket 服务端，
|
||||
覆盖「该重连的」（1011 / 服务器未启动）与「不该重连的」（1000/1001/1002 / 主动断开）两条分支，
|
||||
并断言重连等待取自策略且落在 ±20% 区间、成功后计数清零。
|
||||
|
||||
```
|
||||
|
||||
## 快速开始
|
||||
|
||||
```bash
|
||||
@@ -261,7 +282,7 @@ app/src/main/java/com/opendance/studio/
|
||||
问题大多出在 native 库打包上，而这几项**无设备也能查**：
|
||||
|
||||
```bash
|
||||
python3 build/check_native_libs.py dist-release/opendance-android-v0.2.0/app-release.apk
|
||||
python3 build/check_native_libs.py dist-release/opendance-android-v0.2.1/app-release.apk
|
||||
# -> 每个 ABI 下每个 .so 的 ELF 类型 / DT_NEEDED 闭包 / PT_LOAD 页对齐
|
||||
# -> ✓ native 库体检通过：arm64-v8a, armeabi-v7a ｜ ELF 类型匹配、依赖闭包完整、16 KB 页对齐
|
||||
```
|
||||
@@ -274,7 +295,7 @@ python3 build/check_native_libs.py dist-release/opendance-android-v0.2.0/app-rel
|
||||
而用户装的是 zip 里的 `app-release.apk`。发布前对包再跑一次：
|
||||
|
||||
```bash
|
||||
bash build/accept-release.sh dist-release/opendance-android-v0.2.0.zip
|
||||
bash build/accept-release.sh dist-release/opendance-android-v0.2.1.zip
|
||||
# -> sha256 / 文件清单 / AndroidManifest versionName+versionCode / ABI+MediaPipe native
|
||||
# / 模型资源 / dex 中的 PoseLandmarker / 验签 / 包内契约快照与仓库一致
|
||||
# -> 发布包级验收全部通过
|
||||
@@ -285,13 +306,13 @@ bash build/accept-release.sh dist-release/opendance-android-v0.2.0.zip
|
||||
|
||||
**产物侧复核（T-AND-013）**：`versionName` 是**构建配置**，而安装界面 / 应用信息页 /
|
||||
`dumpsys package` 读的是 **APK 的 `AndroidManifest`**。构建缓存、复用旧 APK、
|
||||
打包时拷错文件都会造成「配置全对、产物是旧版」，配置层校验抓不到（Windows 端
|
||||
v0.2.1 就踩过同类坑）。因此新增：
|
||||
打包时拷错文件都会造成「配置全对、产物是旧版」，配置层校验抓不到（**Windows 客户端**
|
||||
的 v0.2.1 就踩过同类坑：配置文件全一致，exe 属性页显示 0.1.0）。因此新增：
|
||||
|
||||
```bash
|
||||
bash build/check-apk-version.sh app/build/outputs/apk/release/app-release.apk
|
||||
# -> AndroidManifest versionName = 0.2.0 / versionCode = 200
|
||||
# -> ✓ APK 版本元数据一致：0.2.0（versionCode 200）
|
||||
# -> AndroidManifest versionName = 0.2.1 / versionCode = 201
|
||||
# -> ✓ APK 版本元数据一致：0.2.1（versionCode 201）
|
||||
```
|
||||
|
||||
`build/check_apk_version.py` 直接解析二进制 AXML，**不需要 Android SDK / aapt2**
|
||||
|
||||
+15
-2
@@ -1,5 +1,5 @@
|
||||
================================================================
|
||||
OpenDance Studio -- Android 辅助 App v0.2.0 使用说明
|
||||
OpenDance Studio -- Android 辅助 App v0.2.1 使用说明
|
||||
================================================================
|
||||
|
||||
手机端姿态采集器：用摄像头做端侧姿态推理，把 33 个骨骼关键点通过
|
||||
@@ -125,6 +125,19 @@ WebSocket 推送给 Windows 客户端，由后者驱动 VRM 角色并实时评
|
||||
App 运行页 「人体识别」为「已识别」，出现骨架叠加（黄点好、红点置信度低）
|
||||
Windows 端 显示已连接并收到关键点，VRM 模型随动作实时动起来
|
||||
|
||||
6.5) v0.2.1 修订
|
||||
|
||||
1. 重连链路的真实 WebSocket 集成测试（T-AND-016）: 用 MockWebServer 的
|
||||
withWebSocketUpgrade 在 JVM 上起真实 WebSocket 服务端，覆盖「该重连的」
|
||||
（异常关闭码 1011 / 服务端未启动）与「不该重连的」（1000/1001/1002 /
|
||||
用户主动断开）两条分支，并断言重连等待取自策略、落在 ±20% 区间、
|
||||
成功后计数清零。无需设备。
|
||||
2. 修一处真实缺陷: 上一次会话的关闭码会污染下一次会话 —— 第一次被 1001 关闭后
|
||||
用户重新开始推流，若这一次是网络层失败，客户端仍按 1001 判定「不重连」，
|
||||
于是再也不会自动重连，只能反复手点。现在 connect() 会清掉会话级关闭状态，
|
||||
网络失败按「未收到关闭帧(0)」记录并按 §7 重连。
|
||||
3. 单元测试 64 项（原 56 项）。
|
||||
|
||||
7) 自检清单
|
||||
|
||||
./gradlew testDebugUnitTest
|
||||
@@ -144,7 +157,7 @@ WebSocket 推送给 Windows 客户端，由后者驱动 VRM 角色并实时评
|
||||
依赖闭包（DT_NEEDED，非系统库必须同 ABI 自带）、页对齐（16 KB 兼容）。
|
||||
|
||||
0) 发布包级验收（对**发布包**而不是本地构建；文档 §7.5）
|
||||
bash build/accept-release.sh dist-release/opendance-android-v0.2.0.zip
|
||||
bash build/accept-release.sh dist-release/opendance-android-v0.2.1.zip
|
||||
预期末行「发布包级验收全部通过」。它会校验 sha256、解压、核对文件清单、
|
||||
读包内 APK 的 AndroidManifest 版本元数据、核对 ABI 与 MediaPipe native 库、
|
||||
模型资源、dex 里的 PoseLandmarker、验签，以及包内契约快照与仓库是否一致。
|
||||
|
||||
@@ -28,8 +28,8 @@ android {
|
||||
targetSdk = 34
|
||||
// 文档 §11.1：versionCode 用整数按 10000*主 + 100*次 + 修订 递增，
|
||||
// versionName 用 X.Y.Z，两者同一次发布一起更新。
|
||||
versionCode = 200
|
||||
versionName = "0.2.0"
|
||||
versionCode = 201
|
||||
versionName = "0.2.1"
|
||||
|
||||
testInstrumentationRunner = "androidx.test.runner.AndroidJUnitRunner"
|
||||
buildConfigField("String", "WS_SUBPROTOCOL", "\"opendance.pose.v1\"")
|
||||
@@ -225,6 +225,11 @@ dependencies {
|
||||
testImplementation("junit:junit:4.13.2")
|
||||
testImplementation("org.jetbrains.kotlin:kotlin-test:2.0.21")
|
||||
|
||||
// MockWebServer：JVM 上起**真实** WebSocket 服务端（支持 withWebSocketUpgrade），
|
||||
// 用来验证「断线 → 按契约 §7.1 退避重连 / 特定关闭码不重连」这条只在真机上
|
||||
// 才暴露的链路。版本与 okhttp 保持一致，避免协议实现偏斜。
|
||||
testImplementation("com.squareup.okhttp3:mockwebserver:4.12.0")
|
||||
|
||||
// Robolectric：在 JVM 上跑 Android 框架 + Compose，用于验证
|
||||
// 「进入运行页即崩溃」这类只在真机/运行时才暴露的缺陷
|
||||
testImplementation("org.robolectric:robolectric:4.13")
|
||||
|
||||
@@ -146,6 +146,12 @@ class PoseWebSocketClient(
|
||||
this.autoReconnect = autoReconnect
|
||||
reconnectCounter.set(0)
|
||||
lastPongAtMs = 0L
|
||||
// **每次新会话都清掉上一次的关闭码**：`scheduleReconnect()` 是按
|
||||
// `lastCloseCode` 判断「该不该重连」的（契约 §7：1000/1001/1002 不重连），
|
||||
// 若这里不清，上一次的 1001 会把本次会话的**网络失败**也判成「对端主动关闭」，
|
||||
// 于是新版一启动就不再自动重连，用户只能反复手点。
|
||||
// 由 T-AND-016 的 JVM 集成测试（MockWebServer + 真实 WebSocket）复现并锁住。
|
||||
_stats.value = _stats.value.copy(lastCloseCode = 0, lastCloseReason = "")
|
||||
ConnectionLog.record("connect() -> $targetUrl（autoReconnect=$autoReconnect）")
|
||||
val host = runCatching { URI(targetUrl).host }.getOrNull()
|
||||
if (host != null && !isIpLiteral(host)) {
|
||||
@@ -268,6 +274,9 @@ class PoseWebSocketClient(
|
||||
stopHeartbeat()
|
||||
// response != null 时说明 HTTP 升级被拒（例如路径/子协议不对）
|
||||
val httpCode = response?.code ?: 0
|
||||
// 记成 0（未收到关闭帧）：契约 §7 里 0 属于「异常断开 → 应重连」，
|
||||
// 同时让界面「最近一次断开原因」显示的是**本次**失败，而不是上一次的旧值。
|
||||
recordClose(0, if (httpCode > 0) "HTTP $httpCode：$reason" else reason)
|
||||
update(
|
||||
state = State.STOPPED,
|
||||
error = if (httpCode > 0) "HTTP $httpCode：$reason" else reason,
|
||||
|
||||
@@ -0,0 +1,311 @@
|
||||
package com.opendance.studio
|
||||
|
||||
import com.opendance.studio.net.PoseWebSocketClient
|
||||
import com.opendance.studio.net.ReconnectPolicy
|
||||
import com.opendance.studio.util.ConnectionLog
|
||||
import kotlinx.coroutines.CoroutineScope
|
||||
import kotlinx.coroutines.Dispatchers
|
||||
import kotlinx.coroutines.SupervisorJob
|
||||
import kotlinx.coroutines.cancel
|
||||
import okhttp3.OkHttpClient
|
||||
import okhttp3.Response
|
||||
import okhttp3.WebSocket
|
||||
import okhttp3.WebSocketListener
|
||||
import okhttp3.mockwebserver.MockResponse
|
||||
import okhttp3.mockwebserver.MockWebServer
|
||||
import org.junit.After
|
||||
import org.junit.Assert.assertEquals
|
||||
import org.junit.Assert.assertNotNull
|
||||
import org.junit.Assert.assertTrue
|
||||
import org.junit.Before
|
||||
import org.junit.Test
|
||||
import org.junit.runner.RunWith
|
||||
import org.robolectric.RobolectricTestRunner
|
||||
import org.robolectric.annotation.Config
|
||||
import java.util.concurrent.CopyOnWriteArrayList
|
||||
import java.util.concurrent.TimeUnit
|
||||
import java.util.concurrent.atomic.AtomicInteger
|
||||
|
||||
/**
|
||||
* T-AND-016：**重连链路**的 JVM 集成测试（真实 WebSocket，不需要设备）
|
||||
*
|
||||
* 为什么单有 [com.opendance.studio.ReconnectPolicyTest] 不够：
|
||||
* `ReconnectPolicy` 是纯函数，测的是「给定关闭码应该怎么退避」；
|
||||
* 而 ⑦「频繁重连」的缺陷往往出在**调用点**——关闭码是否被正确记录、
|
||||
* 重连计数是否在成功后清零、上一次的关闭码会不会污染下一次会话。
|
||||
* 这些只有把**真实的 OkHttp WebSocket 接到一个真实的服务端**上才能验出来。
|
||||
*
|
||||
* 这里用 MockWebServer 的 `withWebSocketUpgrade` 起真服务端：
|
||||
* * 能按契约 §3.2 回 `welcome`（含 silentTimeoutMs / heartbeatIntervalMs）；
|
||||
* * 能按契约 §3.6 回 `pong`；
|
||||
* * 能用**指定关闭码**关连接（1001 对端离开 / 1011 内部错误 / …），
|
||||
* 从而分别验证「不重连」与「按退避重连」两条分支。
|
||||
*
|
||||
* 注：Robolectric 只用来提供 `android.os.Build.*` 等桩（客户端构造时会读机型名）。
|
||||
*/
|
||||
@RunWith(RobolectricTestRunner::class)
|
||||
@Config(sdk = [34])
|
||||
class PoseWebSocketReconnectTest {
|
||||
|
||||
private lateinit var server: MockWebServer
|
||||
private lateinit var scope: CoroutineScope
|
||||
private val clients = mutableListOf<PoseWebSocketClient>()
|
||||
|
||||
/** 服务端侧收到过的文本消息（hello / ping / stats / bye） */
|
||||
private val serverTexts = CopyOnWriteArrayList<String>()
|
||||
/** 服务端侧见过的连接次数（含重连） */
|
||||
private val serverOpens = AtomicInteger(0)
|
||||
|
||||
private var welcomeSilentTimeoutMs = 4_500L
|
||||
private var closeCodeOnHello = 0
|
||||
private var closeAfterHelloDelayMs = 0L
|
||||
|
||||
@Before
|
||||
fun setUp() {
|
||||
ConnectionLog.clear()
|
||||
serverTexts.clear()
|
||||
serverOpens.set(0)
|
||||
server = MockWebServer()
|
||||
server.start()
|
||||
}
|
||||
|
||||
@After
|
||||
fun tearDown() {
|
||||
clients.forEach { runCatching { it.disconnect() } }
|
||||
scope.cancel()
|
||||
runCatching { server.shutdown() }
|
||||
ConnectionLog.clear()
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ 夹具
|
||||
|
||||
private fun enqueueWebSocket() {
|
||||
server.enqueue(
|
||||
// 必须回显子协议：RFC 6455 要求服务端在 101 响应里带上选中的
|
||||
// Sec-WebSocket-Protocol，客户端据此记录「协商结果」。
|
||||
// （第一版夹具没带，客户端 negotiatedProtocol 就是空串 —— 那是**夹具**不合规，
|
||||
// 不是客户端的问题；这条断言因此改成先在夹具里把服务端做对。）
|
||||
MockResponse()
|
||||
.setHeader("Sec-WebSocket-Protocol", "opendance.pose.v1")
|
||||
.withWebSocketUpgrade(object : WebSocketListener() {
|
||||
override fun onOpen(webSocket: WebSocket, response: Response) {
|
||||
serverOpens.incrementAndGet()
|
||||
}
|
||||
|
||||
override fun onMessage(webSocket: WebSocket, text: String) {
|
||||
serverTexts.add(text)
|
||||
when {
|
||||
text.contains("\"hello\"") -> {
|
||||
webSocket.send(
|
||||
"""{"type":"welcome","protocol":"opendance.pose.v1",""" +
|
||||
""""sessionId":"sess-$serverOpens","expectedFps":30,""" +
|
||||
""""silentTimeoutMs":$welcomeSilentTimeoutMs,""" +
|
||||
""""heartbeatIntervalMs":${ReconnectPolicy.heartbeatIntervalMs(welcomeSilentTimeoutMs)}}""",
|
||||
)
|
||||
if (closeCodeOnHello != 0) {
|
||||
// 模拟「对端主动关闭 / 服务重启」等场景
|
||||
Thread {
|
||||
Thread.sleep(closeAfterHelloDelayMs)
|
||||
webSocket.close(closeCodeOnHello, "test-close-$closeCodeOnHello")
|
||||
}.start()
|
||||
}
|
||||
}
|
||||
text.contains("\"ping\"") -> {
|
||||
webSocket.send("""{"type":"pong","sentAtMs":1,"serverAtMs":2}""")
|
||||
}
|
||||
}
|
||||
}
|
||||
}),
|
||||
)
|
||||
}
|
||||
|
||||
private fun newClient(): PoseWebSocketClient {
|
||||
scope = CoroutineScope(SupervisorJob() + Dispatchers.IO)
|
||||
return PoseWebSocketClient(scope).also { clients.add(it) }
|
||||
}
|
||||
|
||||
private fun wsUrl(): String = server.url("/pose").toString().replace("http://", "ws://")
|
||||
|
||||
/** 轮询等待条件成立（避免依赖固定 sleep 造成的偶发失败） */
|
||||
private fun waitFor(timeoutMs: Long = 6_000, what: String, cond: () -> Boolean): Boolean {
|
||||
val deadline = System.currentTimeMillis() + timeoutMs
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
if (cond()) return true
|
||||
Thread.sleep(25)
|
||||
}
|
||||
throw AssertionError("等待超时（${timeoutMs}ms）：$what")
|
||||
}
|
||||
|
||||
private fun waitUntilOrFalse(timeoutMs: Long, cond: () -> Boolean): Boolean {
|
||||
val deadline = System.currentTimeMillis() + timeoutMs
|
||||
while (System.currentTimeMillis() < deadline) {
|
||||
if (cond()) return true
|
||||
Thread.sleep(25)
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ------------------------------------------------------------------ 用例
|
||||
|
||||
@Test
|
||||
fun `握手成功后进入 streaming 并解析 welcome 的保活参数`() {
|
||||
enqueueWebSocket()
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
|
||||
waitFor(what = "进入 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }
|
||||
|
||||
val stats = client.stats.value
|
||||
assertTrue("sessionId 应来自 welcome（实际 '${stats.serverSessionId}'）", stats.serverSessionId.isNotEmpty())
|
||||
assertEquals("opendance.pose.v1", stats.negotiatedProtocol)
|
||||
assertEquals(welcomeSilentTimeoutMs, stats.silentTimeoutMs)
|
||||
assertEquals(ReconnectPolicy.heartbeatIntervalMs(welcomeSilentTimeoutMs), stats.heartbeatIntervalMs)
|
||||
assertTrue(
|
||||
"hello 必须带 deviceId（契约 §3.1，接收端据此去重）",
|
||||
serverTexts.any { it.contains("\"type\":\"hello\"") && it.contains("\"deviceId\"") },
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `按协商的心跳间隔发送应用层 ping 并收到 pong`() {
|
||||
enqueueWebSocket()
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
waitFor(what = "进入 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }
|
||||
|
||||
waitFor(timeoutMs = 8_000, what = "收到 pong") { client.stats.value.pongsReceived >= 1 }
|
||||
assertTrue(
|
||||
"服务端必须真的收到 ping 文本帧（协议层 ping 对应用层不可见，不能替代）",
|
||||
serverTexts.any { it.contains("\"type\":\"ping\"") },
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `异常关闭码（1011）会按退避策略重连且成功后计数清零`() {
|
||||
closeCodeOnHello = 1011
|
||||
enqueueWebSocket() // 第一次连接：握手后立刻被服务端以 1011 关闭
|
||||
enqueueWebSocket() // 重连后的第二次连接
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
|
||||
// 第一次连接被关闭 → 进入 reconnecting 且尝试次数为 1，等待时间落在策略区间内
|
||||
waitFor(what = "进入 reconnecting") { client.stats.value.state == PoseWebSocketClient.State.RECONNECTING }
|
||||
val firstAttempt = client.stats.value.reconnectAttempts
|
||||
assertEquals(1, firstAttempt)
|
||||
val expected = ReconnectPolicy.baseDelayMs(1) // 500ms
|
||||
val low = (expected * (1 - ReconnectPolicy.JITTER_RATIO)).toLong() // 400ms
|
||||
val high = (expected * (1 + ReconnectPolicy.JITTER_RATIO)).toLong() // 600ms
|
||||
assertTrue(
|
||||
"第 1 次重连等待应落在策略区间 [$low, $high]（实际 ${client.stats.value.nextRetryMs}）",
|
||||
client.stats.value.nextRetryMs in low..high,
|
||||
)
|
||||
|
||||
// 真的重连上了（服务端看到第 2 次 upgrade），并且计数清零、状态回到 streaming
|
||||
waitFor(timeoutMs = 8_000, what = "第二次连接建立") { serverOpens.get() >= 2 }
|
||||
waitFor(what = "重连后回到 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }
|
||||
assertEquals("连接成功后重连计数必须清零（契约 §7.1）", 0, client.stats.value.reconnectAttempts)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `1001 对端离开不重连`() {
|
||||
closeCodeOnHello = 1001
|
||||
enqueueWebSocket()
|
||||
enqueueWebSocket()
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
|
||||
waitFor(what = "进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
|
||||
assertEquals(1001, client.stats.value.lastCloseCode)
|
||||
assertTrue("1001 不应重连（契约 §7）", client.stats.value.reconnectAttempts == 0)
|
||||
val reconnected = waitUntilOrFalse(2_000) { serverOpens.get() >= 2 }
|
||||
assertTrue("1001 之后不允许再建立新连接", !reconnected)
|
||||
assertTrue(
|
||||
"日志里应记录「不重连」便于真机排障",
|
||||
ConnectionLog.entries().any { it.contains("不重连") },
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `1002 协议错误不重连`() {
|
||||
closeCodeOnHello = 1002
|
||||
enqueueWebSocket()
|
||||
enqueueWebSocket()
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
|
||||
waitFor(what = "进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
|
||||
assertEquals(1002, client.stats.value.lastCloseCode)
|
||||
assertTrue("1002 不应重连（重连也不会成功）", !waitUntilOrFalse(2_000) { serverOpens.get() >= 2 })
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `用户主动 disconnect 不重连`() {
|
||||
enqueueWebSocket()
|
||||
enqueueWebSocket()
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
waitFor(what = "进入 streaming") { client.stats.value.state == PoseWebSocketClient.State.STREAMING }
|
||||
|
||||
client.disconnect()
|
||||
waitFor(what = "进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
|
||||
assertTrue("主动断开不应重连", !waitUntilOrFalse(2_000) { serverOpens.get() >= 2 })
|
||||
}
|
||||
|
||||
/**
|
||||
* 本用例是 ⑦ 的关键回归：**上一次会话的关闭码不能污染下一次会话**。
|
||||
*
|
||||
* 复现路径（修复前）：
|
||||
* 1. 第一次连接被对端以 1001 关闭 → lastCloseCode = 1001，按契约不重连（正确）；
|
||||
* 2. 用户再次点「开始推流」→ connect()；
|
||||
* 3. 这一次是**网络层失败**（服务端还没起来 / 被拒），应当按 §7 重连；
|
||||
* 4. 但 `connect()` 没有清理 lastCloseCode，scheduleReconnect() 仍按 1001 判定
|
||||
* → 直接 stopped，**再也不会自动重连**，用户只能反复手点。
|
||||
*/
|
||||
@Test
|
||||
fun `上一次 1001 关闭后重新连接，遇到网络失败仍应按策略重连`() {
|
||||
// 1. 第一次：对端以 1001 关闭
|
||||
closeCodeOnHello = 1001
|
||||
enqueueWebSocket()
|
||||
val client = newClient()
|
||||
client.connect(wsUrl())
|
||||
waitFor(what = "第一次连接进入 stopped") { client.stats.value.state == PoseWebSocketClient.State.STOPPED }
|
||||
assertEquals(1001, client.stats.value.lastCloseCode)
|
||||
|
||||
// 2. 用户重新开始推流，但这一轮服务端 connect 就会失败（直接关掉 MockWebServer）
|
||||
closeCodeOnHello = 0
|
||||
server.shutdown()
|
||||
client.connect(server.url("/pose").toString().replace("http://", "ws://"))
|
||||
|
||||
// 3. 网络失败必须触发重连（进入 reconnecting 或至少 attempts > 0）
|
||||
val recovered = waitUntilOrFalse(6_000) {
|
||||
client.stats.value.state == PoseWebSocketClient.State.RECONNECTING ||
|
||||
client.stats.value.reconnectAttempts > 0
|
||||
}
|
||||
assertTrue(
|
||||
"上一次的 1001 不应影响本次会话：网络失败后必须按 §7 重连" +
|
||||
"（当前 state=${client.stats.value.state}，" +
|
||||
"lastCloseCode=${client.stats.value.lastCloseCode}，" +
|
||||
"attempts=${client.stats.value.reconnectAttempts}）",
|
||||
recovered,
|
||||
)
|
||||
}
|
||||
|
||||
@Test
|
||||
fun `服务端未启动时的连续重连尝试遵守退避（不会疯狂重连）`() {
|
||||
server.shutdown() // 没有服务端
|
||||
val client = newClient()
|
||||
val t0 = System.currentTimeMillis()
|
||||
client.connect(server.url("/pose").toString().replace("http://", "ws://"))
|
||||
|
||||
waitFor(timeoutMs = 6_000, what = "至少安排一次重连") { client.stats.value.reconnectAttempts >= 1 }
|
||||
// 1.5 秒内不应超过 3 次尝试：0.5s → 1s → 2s，说明退避生效
|
||||
Thread.sleep(1_500)
|
||||
val attempts = client.stats.value.reconnectAttempts
|
||||
val elapsed = System.currentTimeMillis() - t0
|
||||
assertTrue(
|
||||
"退避必须生效：${elapsed}ms 内只应有少量尝试（实际 $attempts）",
|
||||
attempts <= 3,
|
||||
)
|
||||
assertNotNull(client.stats.value.lastError)
|
||||
}
|
||||
}
|
||||
+6
-2
@@ -1,5 +1,5 @@
|
||||
project: opendance-android-app
|
||||
version: 0.2.0
|
||||
version: 0.2.1
|
||||
repository: opendance-android
|
||||
default_branch: main
|
||||
platform: Android 10+（minSdk 29 / targetSdk 34）
|
||||
@@ -86,9 +86,13 @@ tasks:
|
||||
file: tasks/015-native-libs.md
|
||||
scope: APK native 库体检（ELF 类型 / DT_NEEDED 闭包 / 16 KB 页对齐），纯标准库实现
|
||||
status: done
|
||||
- id: T-AND-016
|
||||
file: tasks/016-reconnect-integration.md
|
||||
scope: JVM 内以真实 WebSocket（MockWebServer）验证重连链路 + 修「上一次关闭码污染下一次会话」
|
||||
status: done
|
||||
|
||||
deliverable:
|
||||
artifact: opendance-android-v0.2.0.apk
|
||||
artifact: opendance-android-v0.2.1.apk
|
||||
packaging: Gradle 命令行 / Android Studio
|
||||
includes:
|
||||
- app-release.apk
|
||||
|
||||
@@ -0,0 +1,83 @@
|
||||
# T-AND-016：重连链路的 JVM 集成测试（真实 WebSocket）+ 修一处真实缺陷
|
||||
|
||||
- **状态**: done
|
||||
- **依赖**: T-AND-011（连接状态机与重连）
|
||||
- **文档**: v1.6 §6.5 / §7.6
|
||||
- **契约**: `websocket-keypoints 0.2.0` §7 / §7.1 / §8
|
||||
|
||||
## 为什么需要
|
||||
|
||||
`ReconnectPolicyTest`（11 项）验的是**纯函数**「给定关闭码应该怎么退避」；
|
||||
但 ⑦「连接频繁重连」的缺陷往往出在**调用点**：关闭码有没有被正确记录、
|
||||
重连计数有没有在成功后清零、上一次的状态会不会污染下一次会话。
|
||||
这些只有把**真实的 OkHttp WebSocket 接到真实服务端**上才能验出来。
|
||||
|
||||
因此新增 `PoseWebSocketReconnectTest`（8 项）：用 `MockWebServer.withWebSocketUpgrade`
|
||||
在 JVM 上起**真实** WebSocket 服务端（能按 §3.2 回 `welcome`、按 §3.6 回 `pong`、
|
||||
按指定关闭码关连接），无需设备即可覆盖两条分支：**该重连的**与**不该重连的**。
|
||||
Robolectric 只用来提供 `android.os.Build.*` 桩。
|
||||
|
||||
## 它抓到的真实缺陷（本版修复）
|
||||
|
||||
**上一次会话的关闭码会污染下一次会话，导致新版一启动就失去自动重连能力。**
|
||||
|
||||
复现路径：
|
||||
|
||||
1. 第一次连接被对端以 `1001` 关闭 → `lastCloseCode = 1001`，按契约 §7 **不重连**（正确）；
|
||||
2. 用户再次点「开始推流」→ `connect()`；
|
||||
3. 这一次是**网络层失败**（服务端还没起来 / 被拒），按 §7 **应当**重连；
|
||||
4. 但 `connect()` 只重置了 `reconnectCounter` 与 `lastPongAtMs`，**没有清 `lastCloseCode`**；
|
||||
`scheduleReconnect()` 仍按 1001 判定 → 直接 `stopped`，**再也不会自动重连**。
|
||||
|
||||
测试输出的失败信息（修复前）：
|
||||
|
||||
```
|
||||
✗ 上一次 1001 关闭后重新连接，遇到网络失败仍应按策略重连
|
||||
java.lang.AssertionError: 上一次的 1001 不应影响本次会话：网络失败后必须按 §7 重连
|
||||
（当前 state=STOPPED，lastCloseCode=1001，attempts=0）
|
||||
```
|
||||
|
||||
**修复**（`PoseWebSocketClient.kt`）：
|
||||
|
||||
| 位置 | 改动 |
|
||||
|--|--|
|
||||
| `connect()` | 清掉会话级关闭状态：`lastCloseCode = 0`、`lastCloseReason = ""`（每次新会话都从干净状态开始） |
|
||||
| `onFailure()` | 记录 `recordClose(0, …)`（契约 §7 中 0 = 未收到关闭帧 → 应重连），同时让界面「最近一次断开原因」显示**本次**失败而不是旧值 |
|
||||
|
||||
## 用例清单（8 项）
|
||||
|
||||
| 用例 | 验的是什么 |
|
||||
|--|--|
|
||||
| 握手成功 → `streaming`，解析 `welcome` 的 `sessionId` / 保活参数，`hello` 带 `deviceId` | §3.1 / §3.2 发送端一侧 |
|
||||
| 按协商间隔发送应用层 `ping` 并收到 `pong` | §1.1 / §3.6（协议层 ping 不能替代） |
|
||||
| 异常关闭码 `1011` → 按退避重连、等待时间落在策略区间、成功后计数清零 | §7 / §7.1 / §8 |
|
||||
| `1001` 对端离开 → `stopped` 且**不重连**，日志记录「不重连」 | §7 |
|
||||
| `1002` 协议错误 → **不重连** | §7 |
|
||||
| 用户主动 `disconnect()` → **不重连** | §7 |
|
||||
| **上一次 1001 后再连、遇网络失败仍应重连** | 本次修复的缺陷（回归锁） |
|
||||
| 服务端未启动时的连续尝试遵守退避（1.5 s 内 ≤3 次） | §7.1「不会疯狂重连」 |
|
||||
|
||||
## 实测
|
||||
|
||||
```
|
||||
$ ./gradlew :app:testDebugUnitTest
|
||||
BUILD SUCCESSFUL
|
||||
测试类 6 个，用例 64 项，失败 0，跳过 0
|
||||
（新增 8 项；修复前该项失败，其余 7 项通过）
|
||||
```
|
||||
|
||||
`build/verify.sh` 全绿（含签名、版本元数据、native 库体检、ABI、R8、dex）。
|
||||
|
||||
## 踩坑记录
|
||||
|
||||
**夹具也要合规**：第一版 MockWebServer 的 101 响应没有回显
|
||||
`Sec-WebSocket-Protocol`（RFC 6455 要求服务端带上选中的子协议），于是客户端记录的
|
||||
`negotiatedProtocol` 是空串，断言失败。那是**夹具**不合规，不是客户端问题 ——
|
||||
已在夹具里补上 `setHeader("Sec-WebSocket-Protocol", …)`。
|
||||
写这类"真实对端"夹具时，要按 RFC / 契约把服务端做对，否则会把夹具缺陷误判成产品缺陷。
|
||||
|
||||
## 未覆盖（诚实声明）
|
||||
|
||||
- 真机 Wi-Fi 抖动、切网、进程被杀、Doze 省电等**真实断网场景**仍需设备实测；
|
||||
- 契约 §7.1 的**封顶 15 s** 行为由纯函数单测覆盖（穷举 1..8 次），
|
||||
集成测试只验证「等待时间取自策略且落在 ±20% 区间」，否则单测要跑 30 秒以上。
|
||||
Reference in New Issue
Block a user