apps\iot\src\main\java\cn\nfarms\iot\device\DeviceSession.kt¶
源码路径:
apps\iot\src\main\java\cn\nfarms\iot\device\DeviceSession.kt· 389 行
package cn.nfarms.iot.device
import cn.nfarms.iot.ota.OtaPackage
import cn.nfarms.iot.ota.OtaVerdict
import cn.nfarms.iot.ota.OtaVerifier
import cn.nfarms.iot.protocol.Cmd
import cn.nfarms.iot.protocol.Codec
import cn.nfarms.iot.protocol.DeviceStatus
import cn.nfarms.iot.protocol.Frame
import cn.nfarms.iot.transport.BleTransport
import cn.nfarms.iot.transport.Discovered
import cn.nfarms.iot.transport.MockTransport
import java.util.Base64
import java.util.concurrent.TimeoutException
import kotlinx.coroutines.CompletableDeferred
import kotlinx.coroutines.CoroutineScope
import kotlinx.coroutines.Job
import kotlinx.coroutines.delay
import kotlinx.coroutines.flow.MutableStateFlow
import kotlinx.coroutines.flow.StateFlow
import kotlinx.coroutines.flow.asStateFlow
import kotlinx.coroutines.launch
import kotlinx.coroutines.withTimeoutOrNull
/** 连接状态机 —— **显式列出所有状态**,不允许"其他"。 */
enum class LinkState(val label: String) {
Idle("未连接"),
Scanning("扫描中"),
Connecting("连接中"),
Handshaking("握手中"),
Ready("就绪"),
Upgrading("升级中"),
Failed("失败"),
}
data class DeviceUi(
val link: LinkState = LinkState.Idle,
val devices: List<Discovered> = emptyList(),
val status: DeviceStatus = DeviceStatus(),
val transportName: String = "",
val log: List<String> = emptyList(),
val otaProgress: Float = -1f,
val otaVerdict: OtaVerdict? = null,
val message: String? = null,
val rxFrames: Long = 0,
val droppedFrames: Long = 0,
val retries: Int = 0,
)
/**
* 设备会话 = 状态机 + 协议 + 重传。
*
* # 三个设计决定,都是为了在"会丢包"的真实链路上活着
*
* 1. **命令必须有超时**:BLE 丢包是常态(演示传输层默认丢 8%)。
* 没有超时的「write + 等回包」会永久挂住,而界面看起来只是"没反应"。
* 2. **回包用 CompletableDeferred 派发**:收到的帧先解析,能匹配上
* 「当前正在等的那一个」就唤醒它 —— 而不是让每个调用方各自去抢同一条流。
* 这条在 BLE 上很关键:通知是**全局**的,两个同时等待的调用方会互相吃掉对方的包。
* 3. **超时后重试但次数有限**:无限重试会把电量耗在一条断掉的链路上;
* 3 次之后如实报失败,让用户决定。
*/
class DeviceSession(
private val scope: CoroutineScope,
private val transport: BleTransport,
private val verifier: OtaVerifier,
private val deviceTarget: String,
/**
* 连接前就先知道的设备版本。
*
* 为什么需要它:OTA 验签要拿「设备当前版本」比高低,而连上之前它是空的 ——
* 空版本会让**所有**升级包都被判成「版本不高于当前」。演示设备是 1.4.2,
* 预置它之后,连上之前点「升级」也不会被这条误伤。
*/
initialVersion: String = "1.4.2",
) {
private val _ui = MutableStateFlow(
DeviceUi(transportName = transport.name, status = DeviceStatus(version = initialVersion)),
)
val ui: StateFlow<DeviceUi> = _ui.asStateFlow()
/** 当前在等哪一类回包(单飞:同时只允许一个交换在等)。 */
@Volatile private var waiter: CompletableDeferred<FramePayload>? = null
@Volatile private var waiterCmd: Cmd? = null
private data class FramePayload(val cmd: Cmd, val payload: ByteArray)
private val collectJob: Job
private val pollJob: Job
init {
collectJob = scope.launch {
var carry = ByteArray(0)
transport.incoming().collect { chunk ->
val merged = if (carry.isEmpty()) chunk else carry + chunk
val res = Frame.decode(merged)
carry = res.rest
for (d in res.frames) {
if (!d.ok) continue
val mock = transport as? MockTransport
if (mock != null) {
_ui.value = _ui.value.copy(
rxFrames = mock.rxFrames,
droppedFrames = mock.droppedFrames,
)
}
// 只有「正在等的那个命令」才唤醒;其他帧写进日志(不丢,但不阻塞)。
// 这行日志是排障的关键:它能分清「设备没回」和「回了但没认出来」。
log("收到 ${d.cmd.label}(${d.payload.size}B) 等待=${waiterCmd?.label ?: "无"}")
if (d.cmd == waiterCmd) { waiter?.complete(FramePayload(d.cmd, d.payload))
waiter = null
waiterCmd = null
} else {
log("收到未等待的帧:${d.cmd.label}")
}
}
}
}
// 就绪后每 2 秒读一次状态。为什么轮询而不是等推送:
// 固件不一定实现主动上报,而"读一次"永远可行。
pollJob = scope.launch {
while (true) {
delay(2_000)
if (_ui.value.link == LinkState.Ready) runCatching { readStatusNow() }
}
}
}
private fun log(line: String) {
_ui.value = _ui.value.copy(log = (_ui.value.log + line).takeLast(120))
}
// ---------------------------------------------------------------- 连接
fun scan() {
scope.launch {
_ui.value = _ui.value.copy(link = LinkState.Scanning)
val list = runCatching { transport.scan() }.getOrElse {
_ui.value = _ui.value.copy(link = LinkState.Failed, message = "扫描失败:${it.message}")
return@launch
}
_ui.value = _ui.value.copy(devices = list, link = LinkState.Idle)
log("扫描到 ${list.size} 台设备")
}
}
fun connect(deviceId: String) {
scope.launch {
_ui.value = _ui.value.copy(link = LinkState.Connecting)
runCatching { transport.connect(deviceId) }.onFailure {
_ui.value = _ui.value.copy(link = LinkState.Failed, message = "连接失败:${it.message}")
return@launch
}
log("已连接 $deviceId,开始握手")
_ui.value = _ui.value.copy(link = LinkState.Handshaking)
val hello = runCatching { exchange(Cmd.Hello, byteArrayOf(1, 0, 0), Cmd.HelloAck, 3) }.getOrNull()
if (hello == null) {
_ui.value = _ui.value.copy(
link = LinkState.Failed,
message = "握手失败:重试 3 次都没收到 HelloAck(链路在丢包?)",
)
log("握手失败")
return@launch
}
_ui.value = _ui.value.copy(link = LinkState.Ready)
log("握手成功")
readStatusNow()
}
}
fun disconnect() {
scope.launch {
runCatching { transport.disconnect() }
_ui.value = _ui.value.copy(link = LinkState.Idle, status = DeviceStatus())
log("已断开")
}
}
// ---------------------------------------------------------------- 操作
fun readStatus() {
scope.launch { readStatusNow() }
}
private suspend fun readStatusNow() {
val r = runCatching { exchange(Cmd.ReadStatus, ByteArray(0), Cmd.Status, 2) }.getOrNull()
val st = r?.payload?.let(Codec::parseStatus)
if (st == null) {
if (_ui.value.link == LinkState.Ready) log("读状态超时(可能是丢包,下一轮再试)")
return
}
_ui.value = _ui.value.copy(status = st)
if (_ui.value.link == LinkState.Ready) {
log("状态 ${st.temperatureC}°C · ${if (st.power) "开" else "关"} · 模式 ${st.mode} · v${st.version}")
}
}
fun setPower(on: Boolean) {
scope.launch {
val r = runCatching { exchange(Cmd.SetPower, Codec.powerPayload(on), Cmd.Status, 3) }.getOrNull()
val st = r?.payload?.let(Codec::parseStatus)
if (st != null) {
_ui.value = _ui.value.copy(status = st)
log("开关 → ${if (on) "开" else "关"}")
} else {
_ui.value = _ui.value.copy(message = "设置开关失败(重试耗尽)")
log("设置开关失败")
}
}
}
fun setMode(mode: Int) {
scope.launch {
val r = runCatching { exchange(Cmd.SetMode, Codec.modePayload(mode), Cmd.Status, 3) }.getOrNull()
val st = r?.payload?.let(Codec::parseStatus)
if (st != null) {
_ui.value = _ui.value.copy(status = st)
log("模式 → $mode")
} else {
_ui.value = _ui.value.copy(message = "设置模式失败(重试耗尽)")
log("设置模式失败")
}
}
}
// ---------------------------------------------------------------- OTA
/**
* 升级流程:**先验包、再传输、最后生效**。
*
* 验包必须在**下发任何字节之前** —— 把坏包推到设备上再发现不对,
* 轻则白传、重则设备进 bootloader。
*/
fun runOta(pkg: OtaPackage) {
scope.launch {
val wasReady = _ui.value.link == LinkState.Ready
// 入口留一条日志:「点了但什么都没发生」是最难查的一类问题,
// 有这一行就能分清「按钮没调进来」和「进来了但验签/传输失败」。
log("收到升级请求 v${pkg.version} · 当前 v${_ui.value.status.version} · 设备 ${if (wasReady) "就绪" else "未连接"}")
_ui.value = _ui.value.copy(otaVerdict = null, otaProgress = 0f, link = LinkState.Upgrading)
val verdict = runCatching { verifier.verify(pkg, _ui.value.status.version, deviceTarget) }
.getOrElse { OtaVerdict.Rejected("验签异常:${it.message}", OtaVerdict.Stage.Signature) }
_ui.value = _ui.value.copy(otaVerdict = verdict)
if (verdict !is OtaVerdict.Ok) {
val why = (verdict as OtaVerdict.Rejected).why
log("拒绝升级:$why")
_ui.value = _ui.value.copy(
link = if (wasReady) LinkState.Ready else LinkState.Idle,
message = "升级包被拒绝:$why",
)
return@launch
}
if (!wasReady) {
_ui.value = _ui.value.copy(link = LinkState.Idle, message = "先连接设备再升级")
return@launch
}
val bytes = Base64.getDecoder().decode(pkg.payloadB64)
log("验签通过,开始传输 ${bytes.size / 1024} KB")
val began = runCatching { exchange(Cmd.OtaBegin, byteArrayOf(1), Cmd.OtaAck, 3) }.isSuccess
if (!began) {
log("OtaBegin 无应答,中止")
_ui.value = _ui.value.copy(link = LinkState.Ready, message = "升级开始失败")
return@launch
}
val chunk = 64
var idx = 0
var off = 0
while (off < bytes.size) {
val n = minOf(chunk, bytes.size - off)
val payload = Codec.chunkPayload(idx, bytes.copyOfRange(off, off + n))
val ack = runCatching { exchange(Cmd.OtaChunk, payload, Cmd.OtaAck, 4) }.getOrNull()
if (ack == null) {
log("第 $idx 片重试耗尽 —— 中止升级(设备仍在旧固件,不会变砖)")
_ui.value = _ui.value.copy(link = LinkState.Ready, message = "升级中断:第 $idx 片失败")
return@launch
}
off += n
idx++
_ui.value = _ui.value.copy(otaProgress = off.toFloat() / bytes.size)
}
runCatching { exchange(Cmd.OtaEnd, byteArrayOf(1), Cmd.OtaAck, 3) }
_ui.value = _ui.value.copy(otaProgress = 1f, link = LinkState.Ready, message = "升级完成")
log("升级完成,设备版本应为 ${pkg.version}")
delay(400)
readStatusNow()
}
}
// ---------------------------------------------------------------- 交换
/**
* 发一帧、等指定回包。
*
* `tries` 次重试是**由超时驱动的**:每次超时就把帧再发一遍。
* 注意重发前要先把 waiter 清掉,否则上一轮的回包会唤醒下一轮的等待
* (这种"错位的回包"在丢包链路上非常容易发生)。
*
* # 计时必须从**写完**开始
*
* 这里踩过一个真坑:一开始写成「先建 waiter,再 write(把超时套在外面)」。
* 在模拟器上,写一个 68 字节的 OtaChunk 要按 MTU 分 4 段、每段还带调度抖动,
* 于是**首包的 800ms 预算在发出最后一字节之前就被吃掉了** ——
* 设备明明回了 ack,会话这边已经超时重发,日志里只剩一串
* 「等升级应答超时,重发」。桌面 JVM 上时序快,同样的代码却是绿的。
*
* 正确的顺序:write 完 → 建 waiter → 才开始计时。
*/
private suspend fun exchange(
cmd: Cmd,
payload: ByteArray,
expect: Cmd,
tries: Int,
): FramePayload {
var lastErr: Throwable? = null
repeat(tries) { attempt ->
val frame = Frame.encode(cmd, payload)
// **先把 waiter 立起来,再发**。
//
// 反过来(先 write 再建 waiter)会造成一个真竞态:模拟器上设备的回包
// 可能在 `write` 返回之前就通过 SharedFlow 到达,那时 `waiterCmd` 还是 null,
// 于是这一帧被判成「未等待的帧」直接丢掉 —— 日志里就是
// 收到未等待的帧:握手应答 / 等 握手应答 超时,重发(1/3)
// 桌面 JVM 上时序快慢不同,同样的代码不会暴露。
val deferred = CompletableDeferred<FramePayload>()
waiter = deferred
waiterCmd = expect
try {
transport.write(frame)
// 计时**从写完开始**:分片传输本身花的时间不该吃掉等回包的预算
// (68 字节的 OtaChunk 在模拟器上光写就要 30ms 以上)。
// 但 waiter 早已就位,所以这段时间里到达的回包一条都不会丢。
val got = withTimeoutOrNull(timeoutFor(frame.size)) { deferred.await() }
if (got != null) return got
lastErr = TimeoutException("等 ${expect.label} 超时")
if (attempt > 0) {
_ui.value = _ui.value.copy(retries = _ui.value.retries + 1)
}
log("等 ${expect.label} 超时,重发(${attempt + 1}/$tries)")
} finally {
if (waiter === deferred) {
waiter = null
waiterCmd = null
}
}
delay(60)
}
throw lastErr ?: IllegalStateException("交换失败")
}
fun dismissMessage() { _ui.value = _ui.value.copy(message = null) }
fun close() {
collectJob.cancel()
pollJob.cancel()
}
companion object {
/**
* 基准等待上限。
*
* 1.5s:BLE 一次往返通常在 30~120ms,但**模拟器/低端机的 UI 线程一忙**,
* 200ms 级别的等待会偶发超时(表现为"回了 ack 却报超时")。给到 1.5s
* 仍有「连续丢 5 个包」的余量,而真机上的失败感知只慢 0.7s,可接受。
*
* 见 [timeoutFor]:大帧会按 MTU 分片,等待上限要相应放大。
*/
const val TIMEOUT_MS: Long = 1_500
/** 一个 MTU 能装多少(真机默认 23,其中 3 字节是 ATT 头)。 */
private const val MTU = 20
/**
* 按帧大小算等待上限。
*
* 一帧要被切成 `ceil(size / MTU)` 片;设备是在**收齐**之后才回包,
* 所以等待时间必须覆盖「收齐 + 回包」,否则大帧永远超时。
* 每片按 15ms 估(8ms 的模拟写间隔 + 调度抖动),再加基准 800ms。
*/
fun timeoutFor(frameSize: Int): Long {
val pieces = (frameSize + MTU - 1) / MTU
return TIMEOUT_MS + pieces * 15L
}
}
}