From 1bd57e068043b5c92d78b67047982b54368a6374 Mon Sep 17 00:00:00 2001 From: mazengfei <331023091@qq.com> Date: Wed, 2 Sep 2026 14:57:35 +0800 Subject: [PATCH] =?UTF-8?q?feat(mqtt):=20=E5=A2=9E=E5=8A=A0=E4=BA=BA?= =?UTF-8?q?=E8=84=B8=E5=BA=93=20MQTT=20=E5=AE=9E=E6=97=B6=E5=8F=98?= =?UTF-8?q?=E6=9B=B4=E5=90=8C=E6=AD=A5=E5=8A=9F=E8=83=BD?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - 新增 MQTT 客户端依赖及配置，实现设备端实时接收人脸变化推送 - 实现 FaceMqttSubscriber 单例，启动时解析环境选择对应 MQTT 配置 - 实现消息解析、乱序过滤及本地人脸库数据实时更新逻辑 - 设备端订阅人脸变更主题，连接/重连成功触发 HTTP 增量补拉兜底机制 - 在 LoginByFaceActivity 中订阅 MQTT 事件，实时刷新识别引擎内存 - MQTT 连接管理器 MqttManager 负责连接、重连、订阅及消息分发，实现完整生命周期管理 - 新增环境选择弹窗，优化初始化界面布局和按钮样式 - 配置激活码硬编码修正，避免测试环境激活码错误导致的问题 - 日志与错误提示等细节优化，提升稳定性与用户体验 --- app/build.gradle.kts | 3 + .../main/java/com/sw/platecabinet/MyApp.kt | 4 + .../activity/DeviceInitActivity.kt | 4 +- .../activity/LoginByFaceActivity.kt | 31 ++ .../dialog/EnvironmentSelectDialog.kt | 20 +- .../com/sw/platecabinet/mqtt/FaceMqttEvent.kt | 7 + .../platecabinet/mqtt/FaceMqttSubscriber.kt | 281 ++++++++++++++++++ .../com/sw/platecabinet/mqtt/MqttConfig.kt | 30 ++ .../com/sw/platecabinet/mqtt/MqttManager.kt | 211 +++++++++++++ .../com/sw/platecabinet/mqtt/MqttMessage.kt | 11 + .../platecabinet/viewmodel/NetViewModelV2.kt | 2 +- app/src/main/res/layout/activity_init.xml | 48 ++- .../res/layout/dialog_environment_select.xml | 36 +++ app/src/main/res/values/colors.xml | 1 + app/src/main/res/values/strings.xml | 1 + gradle/libs.versions.toml | 2 + .../utils/arcface/FaceRectTransformer.java | 1 - .../plate/utils/arcface/face/FaceHelper.java | 1 - 18 files changed, 658 insertions(+), 36 deletions(-) create mode 100644 app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttEvent.kt create mode 100644 app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt create mode 100644 app/src/main/java/com/sw/platecabinet/mqtt/MqttConfig.kt create mode 100644 app/src/main/java/com/sw/platecabinet/mqtt/MqttManager.kt create mode 100644 app/src/main/java/com/sw/platecabinet/mqtt/MqttMessage.kt create mode 100644 app/src/main/res/layout/dialog_environment_select.xml diff --git a/app/build.gradle.kts b/app/build.gradle.kts index dd996cf..256c7c3 100644 --- a/app/build.gradle.kts +++ b/app/build.gradle.kts @@ -100,4 +100,7 @@ dependencies { implementation(libs.android.core) implementation("org.greenrobot:eventbus:3.3.1") + + // MQTT 客户端（人脸变更实时推送） + implementation(libs.paho.mqtt) } \ No newline at end of file diff --git a/app/src/main/java/com/sw/platecabinet/MyApp.kt b/app/src/main/java/com/sw/platecabinet/MyApp.kt index 3a6718a..321ebc5 100644 --- a/app/src/main/java/com/sw/platecabinet/MyApp.kt +++ b/app/src/main/java/com/sw/platecabinet/MyApp.kt @@ -3,6 +3,7 @@ package com.sw.platecabinet import android.util.Log import com.sw.plate.App import com.sw.plate.utils.AppUtil +import com.sw.platecabinet.mqtt.FaceMqttSubscriber import com.sw.platecabinet.utils.CrashHandler import com.sw.platecabinet.utils.SpTool import timber.log.Timber @@ -20,6 +21,9 @@ class MyApp : App() { // 初始化崩溃处理器 CrashHandler.init(this) + + // 启动人脸 MQTT 实时订阅（进程级单例，应用生命周期内保持连接） + FaceMqttSubscriber.start() } /** diff --git a/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt b/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt index a5ebd55..b56f864 100644 --- a/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt +++ b/app/src/main/java/com/sw/platecabinet/activity/DeviceInitActivity.kt @@ -48,7 +48,9 @@ class DeviceInitActivity : BaseActivity() { } GlobalData.appId = deviceConfig.arcsoftAppId ?: "" GlobalData.sdkKey = deviceConfig.arcsoftSdkKey ?: "" - GlobalData.activeKey = deviceConfig.arcsoftActiveKey ?: "" + // 测试设备后台下发的激活码不正确，手动硬编码覆盖为正确值（后台修复后可回退为下行） + // GlobalData.activeKey = deviceConfig.arcsoftActiveKey ?: "" + GlobalData.activeKey = "085F-118G-Q4V1-THBP" hideWaitingDialog() goLoginActivity() } diff --git a/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt b/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt index b6dd520..e93a2f7 100644 --- a/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt +++ b/app/src/main/java/com/sw/platecabinet/activity/LoginByFaceActivity.kt @@ -35,6 +35,8 @@ import com.sw.platecabinet.member.R import com.sw.platecabinet.member.databinding.ActivityLoginFaceBinding import com.sw.platecabinet.member.databinding.ItemTitleTimeBinding import com.sw.platecabinet.model.response.EquipmentUserInfo +import com.sw.platecabinet.mqtt.FaceChangedEvent +import com.sw.platecabinet.mqtt.FaceSyncTriggerEvent import com.sw.platecabinet.network.task.TaskManager import com.sw.platecabinet.utils.PermissionHelper import org.greenrobot.eventbus.EventBus @@ -143,6 +145,35 @@ class LoginByFaceActivity : BaseActivity(), Timber.tag("performSync over").e("-time=%s", insertEntity.registerTime) } + /** + * MQTT 实时同步后的人脸库已落库，防抖重载识别引擎内存（合并短时间内连续变更） + */ + private val faceRefreshHandler = Handler(Looper.getMainLooper()) + private val faceRefreshRunnable = Runnable { recognizeViewModel.refreshFaceList() } + + @Subscribe(threadMode = ThreadMode.MAIN) + fun onFaceChanged(event: FaceChangedEvent) { + Timber.d("onFaceChanged 收到人脸实时变更，2s 后刷新引擎内存") + faceRefreshHandler.removeCallbacks(faceRefreshRunnable) + faceRefreshHandler.postDelayed(faceRefreshRunnable, 2000) + } + + /** + * MQTT 连接/重连成功，补拉一次 HTTP 增量兜底（设备离线期间漏收的变更） + */ + @Subscribe(threadMode = ThreadMode.MAIN) + fun onFaceSyncTrigger(event: FaceSyncTriggerEvent) { + val timestamp = SpTool.getLastFaceTimestamp() + if (timestamp == 0L) { + Timber.d("onFaceSyncTrigger 水位为 0，跳过增量补拉") + return + } + Timber.d("onFaceSyncTrigger MQTT 已连接，执行 HTTP 增量补拉 timestamp=$timestamp") + netViewModelV2.getFaceIncrementList(pageNo = 1, timestamp = timestamp) { + recognizeViewModel.refreshFaceList() + } + } + override fun onLeftDoubleClick() { finish() } diff --git a/app/src/main/java/com/sw/platecabinet/dialog/EnvironmentSelectDialog.kt b/app/src/main/java/com/sw/platecabinet/dialog/EnvironmentSelectDialog.kt index 11cffcd..855ab6f 100644 --- a/app/src/main/java/com/sw/platecabinet/dialog/EnvironmentSelectDialog.kt +++ b/app/src/main/java/com/sw/platecabinet/dialog/EnvironmentSelectDialog.kt @@ -5,6 +5,10 @@ import android.content.res.ColorStateList import android.graphics.Color import android.graphics.drawable.ColorDrawable import android.os.Bundle +import android.text.SpannableString +import android.text.Spanned +import android.text.style.AbsoluteSizeSpan +import android.text.style.ForegroundColorSpan import android.view.Gravity import android.view.View import android.view.Window @@ -50,9 +54,8 @@ class EnvironmentSelectDialog( ENVIRONMENTS.forEach { env -> val rb = RadioButton(activity).apply { id = View.generateViewId() - text = env.name + text = buildEnvironmentLabel(env.name, env.url) gravity = Gravity.CENTER_VERTICAL - textSize = 18f setTextColor(Color.WHITE) buttonTintList = ColorStateList.valueOf(Color.WHITE) isChecked = env.url == current @@ -69,6 +72,19 @@ class EnvironmentSelectDialog( selected = ENVIRONMENTS.firstOrNull { it.url == current } } + /** + * 构造环境选项文本：名称（白色 18sp）+ 换行 + 地址（灰色 13sp） + */ + private fun buildEnvironmentLabel(name: String, url: String): SpannableString { + val label = "$name\n$url" + val spannable = SpannableString(label) + val urlStart = name.length + 1 // 跳过换行符 + spannable.setSpan(AbsoluteSizeSpan(18, true), 0, name.length, Spanned.SPAN_EXCLUSIVE_EXCLUSIVE) + spannable.setSpan(AbsoluteSizeSpan(13, true), urlStart, label.length, Spanned.SPAN_EXCLUSIVE_EXCLUSIVE) + spannable.setSpan(ForegroundColorSpan(Color.GRAY), urlStart, label.length, Spanned.SPAN_EXCLUSIVE_EXCLUSIVE) + return spannable + } + /** * 确认按钮：回调所选环境并关闭弹窗 */ diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttEvent.kt b/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttEvent.kt new file mode 100644 index 0000000..b469823 --- /dev/null +++ b/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttEvent.kt @@ -0,0 +1,7 @@ +package com.sw.platecabinet.mqtt + +/** 人脸库已由 MQTT 实时更新（DB 已落库），通知界面层刷新识别引擎内存 */ +class FaceChangedEvent + +/** MQTT 连接/重连成功，请求执行一次 HTTP 增量补拉兜底 */ +class FaceSyncTriggerEvent diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt b/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt new file mode 100644 index 0000000..c4fd673 --- /dev/null +++ b/app/src/main/java/com/sw/platecabinet/mqtt/FaceMqttSubscriber.kt @@ -0,0 +1,281 @@ +package com.sw.platecabinet.mqtt + +import com.google.gson.JsonParser +import com.sw.plate.utils.arcface.FaceApi +import com.sw.plate.utils.arcface.facedb.entity.FaceEntity +import com.sw.platecabinet.GlobalData +import com.sw.platecabinet.model.response.UserFaceModelV2 +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.launch +import org.greenrobot.eventbus.EventBus +import timber.log.Timber + +/** + * 人脸 MQTT 实时同步订阅器（进程级单例，替代 Hilt 注入） + * + * 订阅服务端人脸变更广播主题（随环境区分）：用户在小程序/管理端采集或更换人脸后， + * 服务端实时推送特征码变更，设备端立即更新本地人脸库，无需等待 5 分钟定时增量轮询。 + * + * 与增量接口的联动：MQTT 提供秒级实时推送；设备侧另有 5 分钟定时轮询 + + * 连接/重连成功时补拉一次 HTTP 增量，兜底推送失败与设备离线 + * （关机/断网/broker 会话过期）漏收的变更。增量水位只由 HTTP 响应推进， + * 本模块不改动水位。 + * + * 幂等与乱序按 userFaceId + 时间戳守卫处理（见 [applyRealtimeUpdates]）。 + */ +object FaceMqttSubscriber { + + private const val TAG = "FaceMqttSubscriber" + + /** MQTT 账号（dev 域名与测试 IP 共用） */ + private const val MQTT_USER = "platform" + private const val MQTT_PASSWORD = "ZrKhZhng6t2tlpid" + + /** MQTT 账号（UAT 独立） */ + private const val MQTT_USER_UAT = "platform-uat" + private const val MQTT_PASSWORD_UAT = "p2AXu3lsUllbXJEY_A1!" + + /** + * 根据业务服务器 BASE_URL 解析对应的 MQTT 环境配置（host、port、协议、订阅主题）。 + * 与当前项目的三个预设环境对齐：本地 / 测试 / 生产(UAT)。 + * 未匹配的环境返回 null 表示不接入实时同步。 + */ + private fun resolveMqttEnv(baseUrl: String): MqttEnv? = when { + baseUrl.startsWith("https://platform-api.uat.shuziweidao.com") -> + MqttEnv("mqtt.uat.shuziweidao.com", 443, "wss", "yx/device/face/update-local", + MQTT_USER_UAT, MQTT_PASSWORD_UAT) + baseUrl.startsWith("https://dev.yixiong-tech.com") -> + MqttEnv("dev.yixiong-tech.com", 8089, "wss", "yx/device/face/update-test", + MQTT_USER, MQTT_PASSWORD) + baseUrl.startsWith("http://192.168.10.101") -> + MqttEnv("192.168.10.101", 1884, "tcp", "yx/device/face/update-dev", + MQTT_USER, MQTT_PASSWORD) + else -> null + } + + /** MQTT 环境配置：host、port、协议、订阅主题、账号 */ + private data class MqttEnv( + val host: String, + val port: Int, + val scheme: String, + val topic: String, + val userName: String, + val password: String + ) + + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + private val faceApi = FaceApi() + + private var started = false + private var collectJob: Job? = null + private var stateJob: Job? = null + + /** 当前环境的人脸变更订阅主题（随 start() 按 baseUrl 环境解析） */ + private var faceUpdateTopic: String = "" + + /** + * 启动订阅（幂等，重复调用不会建立多条连接） + * + * clientId 使用 `platecabinet-{设备UDID}`，与 X-DEVICE-CODE 同源，保证一台设备一条连接。 + * 应在应用启动、设备初始化完成后调用。 + */ + fun start() { + if (started) return + started = true + + val baseUrl = GlobalData.appBaseUrl + val env = resolveMqttEnv(baseUrl) ?: run { + started = false + Timber.w("$TAG 当前环境（baseUrl=$baseUrl）无 MQTT 配置，跳过实时同步，人脸变更仅靠定时增量轮询兜底") + return + } + faceUpdateTopic = env.topic + val udid = GlobalData.deviceId.ifBlank { "unknown" } + Timber.i( + "$TAG 启动人脸 MQTT 订阅, broker=${env.scheme}://${env.host}:${env.port}, " + + "topic=${env.topic}, clientId=platecabinet-$udid" + ) + + MqttManager.configure( + MqttConfig( + host = env.host, + port = env.port, + scheme = env.scheme, + clientId = "platecabinet-$udid", + userName = env.userName, + password = env.password, + cleanSession = false + ) + ) + MqttManager.connect() + + // 消费广播消息：解析 → 落库 → 通知界面刷新引擎内存 + collectJob = scope.launch { + MqttManager.messages.collect { message -> + if (message is MqttMessage.Received) { + if (message.topic == faceUpdateTopic) { + handleFaceUpdate(message.payload) + } else { + // 联调排查：后台发送的 topic 与预期不一致时在此可见 + Timber.w("$TAG 收到非人脸主题消息，忽略: topic=${message.topic} size=${message.payload.size}") + } + } + } + } + + // 连接状态联动：连接成功 → 恢复订阅 + 触发一次 HTTP 增量补拉 + stateJob = scope.launch { + MqttManager.state.collect { state -> + if (state == MqttState.Connected) { + MqttManager.subscribe(faceUpdateTopic, 1) + // 设备离线期间（关机/断网/会话过期）的变更 MQTT 无法补收， + // 通知界面层用水位时间戳补拉一次 HTTP 增量兜底 + EventBus.getDefault().post(FaceSyncTriggerEvent()) + } + } + } + } + + fun stop() { + if (!started) return + started = false + collectJob?.cancel() + stateJob?.cancel() + collectJob = null + stateJob = null + MqttManager.disconnect() + } + + /** 解析广播 JSON 数组并落库（faceUpdateTimestamp 兼容 long/String 两种格式） */ + private fun handleFaceUpdate(payload: ByteArray) { + // 联调排查：记录原始报文预览（特征码很长，只打前 120 字符） + val raw = String(payload, Charsets.UTF_8) + Timber.i("$TAG 收到人脸广播: size=${payload.size} 预览=${raw.take(120)}") + val items = try { + parsePayload(raw) + } catch (e: Exception) { + Timber.e(e, "$TAG 广播消息解析失败: ${raw.take(300)}") + return + } + // 联调排查：逐条记录关键字段（最多 5 条，特征码只打长度） + items.take(5).forEach { + Timber.i( + "$TAG 解析条目: userFaceId=${it.userFaceId} userId=${it.userId} " + + "deleted=${it.faceDeleted} ts=${it.faceUpdateTimestamp} " + + "featureLen=${it.resolveFeatureStr()?.length ?: 0}" + ) + } + Timber.i("$TAG 收到人脸广播 ${items.size} 条，开始落库") + applyRealtimeUpdates(items) + Timber.i("$TAG 人脸广播处理完成 ${items.size} 条") + } + + /** 解析 JSON 数组为 [UserFaceModelV2]（时间戳兼容 long 数字与字符串两种格式） */ + private fun parsePayload(raw: String): List { + val array = JsonParser.parseString(raw).asJsonArray + return array.mapNotNull { element -> + val obj = element.asJsonObject + val tsElement = obj.get("faceUpdateTimestamp") + val timestamp = when { + tsElement == null || tsElement.isJsonNull -> null + tsElement.isJsonPrimitive -> tsElement.asString.toLongOrNull() + else -> null + } + UserFaceModelV2( + userFaceId = obj.get("userFaceId")?.takeIf { !it.isJsonNull }?.asString, + userId = obj.get("userId")?.takeIf { !it.isJsonNull }?.asString, + faceFeature = obj.get("faceFeature")?.takeIf { !it.isJsonNull }?.asString, + faceFeatureStr = null, + faceFeatureString = null, + faceUpdateTimestamp = timestamp, + cardNo = obj.get("cardNo")?.takeIf { !it.isJsonNull }?.asString, + member = obj.get("member")?.takeIf { !it.isJsonNull }?.asBoolean, + faceDeleted = obj.get("faceDeleted")?.takeIf { !it.isJsonNull }?.asBoolean ?: false, + personType = obj.get("personType")?.takeIf { !it.isJsonNull }?.asString + ) + } + } + + /** + * 应用 MQTT 实时推送的人脸变更（立即落库，不等下一轮轮询） + * + * 处理语义与增量接口一致，另加乱序守卫： + * 按 userFaceId 查本地记录，消息时间戳不大于本地时间戳的条目直接丢弃， + * 防止 QoS1 重复投递或乱序到达时旧数据覆盖新数据。 + * + * 注意：本方法不推进增量水位。水位只由 HTTP 增量接口的响应推进， + * 后台推送失败只记日志不重发，若 MQ 消息把水位推到漏发变更之后， + * 轮询/补拉将永远拉不到那条变更。 + */ + private fun applyRealtimeUpdates(items: List) { + if (items.isEmpty()) return + + // 批内去重：同一 userFaceId 仅保留时间戳最大的一条（含其删除标志） + val batch = items.groupBy { it.userFaceId } + .flatMap { (key, list) -> + if (key.isNullOrEmpty()) list + else listOfNotNull(list.maxByOrNull { it.faceUpdateTimestamp ?: 0L }) + } + + // 乱序守卫：本地已有同 userFaceId 且时间戳不旧的记录则跳过该条 + val fresh = batch.filter { msg -> + val key = msg.userFaceId + if (key.isNullOrEmpty()) return@filter true + val local = faceApi.queryByUserFaceId(key) + val msgTs = msg.faceUpdateTimestamp ?: 0L + when { + local == null -> true + msgTs <= 0L -> true + msgTs > local.faceUpdateTimestamp -> true + else -> { + Timber.d("$TAG 实时消息乱序/重复，丢弃 userFaceId=$key ts=$msgTs") + false + } + } + } + if (fresh.isEmpty()) return + + var changed = false + for (msg in fresh) { + if (msg.faceDeleted == true) { + val userFaceId = msg.userFaceId + if (!userFaceId.isNullOrEmpty() && faceApi.queryByUserFaceId(userFaceId) != null) { + faceApi.deleteByUserFaceId(userFaceId) + changed = true + } + } else { + // 优先按 userFaceId 精确判重，避免特征字段不一致导致重复入库 + val userFaceId = msg.userFaceId + if (!userFaceId.isNullOrEmpty() && faceApi.queryByUserFaceId(userFaceId) != null) { + continue + } + val entity = buildEntity(msg) ?: continue + faceApi.insert(entity) + changed = true + } + } + + if (changed) { + Timber.i("$TAG MQTT 实时更新完成 ${items.size} 条（实际变更 $changed 条），通知界面刷新引擎") + EventBus.getDefault().post(FaceChangedEvent()) + } + } + + /** 将 MQTT 消息转换为 [FaceEntity]，特征 Base64 为空或解码失败返回 null */ + private fun buildEntity(model: UserFaceModelV2): FaceEntity? { + val featureBase64 = model.resolveFeatureStr() + if (featureBase64.isNullOrBlank()) { + Timber.w("$TAG 人脸特征为空，跳过入库: userFaceId=${model.userFaceId} userId=${model.userId}") + return null + } + return try { + model.toFaceEntity() + } catch (e: Exception) { + Timber.e(e, "$TAG 人脸特征 Base64 解码失败: userFaceId=${model.userFaceId}") + null + } + } +} diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/MqttConfig.kt b/app/src/main/java/com/sw/platecabinet/mqtt/MqttConfig.kt new file mode 100644 index 0000000..65d18ff --- /dev/null +++ b/app/src/main/java/com/sw/platecabinet/mqtt/MqttConfig.kt @@ -0,0 +1,30 @@ +package com.sw.platecabinet.mqtt + +/** + * MQTT 连接配置 + * + * @param host Broker 主机地址 + * @param port Broker 端口，默认 1883 + * @param scheme 连接协议：tcp（裸 MQTT）、ws（MQTT over WebSocket）、wss（WebSocket + TLS）。 + * EMQX 惯例 1883=tcp / 8083=ws / 8084=wss，端口与协议不匹配时 Broker 会在 + * CONNACK 前直接断开（32109/EOFException） + * @param clientId 客户端唯一标识；同一 Broker 下多台设备禁止重复，否则旧连接会被踢下线 + * @param userName 鉴权用户名，null 表示匿名连接 + * @param password 鉴权密码，null 表示无密码 + * @param cleanSession false 时保留会话（离线期间 QoS1 消息由 Broker 暂存，重连后补收） + * @param keepAliveInterval 心跳间隔（秒） + * @param connectionTimeout 连接超时（秒） + */ +data class MqttConfig( + val host: String, + val port: Int = 1883, + val scheme: String = "tcp", + val clientId: String? = null, + val userName: String? = null, + val password: String? = null, + val cleanSession: Boolean = true, + val keepAliveInterval: Int = 20, + val connectionTimeout: Int = 30 +) { + val brokerUrl: String get() = "$scheme://$host:$port" +} diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/MqttManager.kt b/app/src/main/java/com/sw/platecabinet/mqtt/MqttManager.kt new file mode 100644 index 0000000..b5d6228 --- /dev/null +++ b/app/src/main/java/com/sw/platecabinet/mqtt/MqttManager.kt @@ -0,0 +1,211 @@ +package com.sw.platecabinet.mqtt + +import kotlinx.coroutines.CoroutineScope +import kotlinx.coroutines.Dispatchers +import kotlinx.coroutines.Job +import kotlinx.coroutines.SupervisorJob +import kotlinx.coroutines.cancel +import kotlinx.coroutines.delay +import kotlinx.coroutines.flow.MutableSharedFlow +import kotlinx.coroutines.flow.MutableStateFlow +import kotlinx.coroutines.flow.SharedFlow +import kotlinx.coroutines.flow.StateFlow +import kotlinx.coroutines.flow.asSharedFlow +import kotlinx.coroutines.flow.asStateFlow +import kotlinx.coroutines.launch +import org.eclipse.paho.client.mqttv3.IMqttDeliveryToken +import org.eclipse.paho.client.mqttv3.MqttCallbackExtended +import org.eclipse.paho.client.mqttv3.MqttClient +import org.eclipse.paho.client.mqttv3.MqttConnectOptions +import org.eclipse.paho.client.mqttv3.persist.MemoryPersistence +import timber.log.Timber + +/** MQTT 连接状态 */ +enum class MqttState { Disconnected, Connecting, Connected } + +/** + * MQTT 连接管理器（进程级单例，替代 Hilt 注入） + * + * 负责 Broker 连接、自动重连、订阅管理与消息分发： + * - 启用 Paho 自动重连，重连成功后自动恢复历史订阅（[MqttCallbackExtended.connectComplete]） + * - 业务消息通过 [messages] SharedFlow 分发，业务层按 topic 过滤消费 + * - cleanSession=false 时 Broker 会暂存离线期间的 QoS1 消息，重连后补收 + */ +object MqttManager { + private val scope = CoroutineScope(SupervisorJob() + Dispatchers.IO) + private var client: MqttClient? = null + private var config: MqttConfig? = null + + /** 已订阅的 topic 及 QoS，自动重连成功后按此恢复订阅 */ + private val subscriptions = linkedMapOf() + + private val _state = MutableStateFlow(MqttState.Disconnected) + val state: StateFlow = _state.asStateFlow() + + private val _messages = MutableSharedFlow(extraBufferCapacity = 20) + val messages: SharedFlow = _messages.asSharedFlow() + + /** 首连失败后的退避重试协程，连接成功或主动断开时取消 */ + private var retryJob: Job? = null + + fun configure(cfg: MqttConfig) { + config = cfg + Timber.d("MQTT 配置完成: ${cfg.brokerUrl}") + } + + fun connect() { + val cfg = config ?: run { + Timber.e("MQTT 未配置，请先调用 configure()") + return + } + // 已连接或正在连接时重复调用直接忽略，避免重建客户端导致重复连接 + if (_state.value != MqttState.Disconnected) { + Timber.d("MQTT 当前状态=${_state.value}，忽略重复 connect()") + return + } + retryJob?.cancel() + retryJob = scope.launch { doConnect(cfg) } + } + + /** + * 建立连接，失败则指数退避重试 + * + * Paho 的 automaticReconnect 仅在首连成功后生效；设备开机时 Broker 可能 + * 尚未就绪导致首连失败，故此处自建退避循环兜底（1s 起、翻倍、60s 封顶）。 + */ + private suspend fun doConnect(cfg: MqttConfig) { + val id = cfg.clientId ?: "platecabinet_${System.currentTimeMillis()}" + var backoffMs = 1_000L + while (true) { + try { + _state.value = MqttState.Connecting + client = MqttClient(cfg.brokerUrl, id, MemoryPersistence()).apply { + setCallback(createCallback()) + val options = MqttConnectOptions().apply { + isCleanSession = cfg.cleanSession + keepAliveInterval = cfg.keepAliveInterval + connectionTimeout = cfg.connectionTimeout + isAutomaticReconnect = true + maxReconnectDelay = 60_000 + // 鉴权信息（用户名非空时才注入，避免空串覆盖匿名连接） + cfg.userName?.let { userName = it } + cfg.password?.let { password = it.toCharArray() } + } + connect(options) + } + _state.value = MqttState.Connected + Timber.i("MQTT 连接成功: ${cfg.brokerUrl}, clientId=$id, cleanSession=${cfg.cleanSession}") + return + } catch (e: Exception) { + // 连接失败时 MqttClient 可能残留半初始化状态，先关闭清理 + try { client?.close() } catch (_: Exception) {} + client = null + _state.value = MqttState.Disconnected + _messages.emit(MqttMessage.Error("连接失败: ${e.message}", e)) + Timber.e(e, "MQTT 连接失败，${backoffMs / 1000} 秒后重试") + delay(backoffMs) + backoffMs = (backoffMs * 2).coerceAtMost(60_000L) + } + } + } + + fun disconnect() { + // 取消首连重试循环，避免主动断开后仍反复重连 + retryJob?.cancel() + retryJob = null + scope.launch { + try { client?.disconnect() } catch (_: Exception) {} + client?.close() + client = null + _state.value = MqttState.Disconnected + _messages.emit(MqttMessage.Disconnected()) + Timber.d("MQTT 已断开") + } + } + + /** 订阅主题（记录 QoS，自动重连成功后由 [createCallback] 恢复） */ + fun subscribe(topic: String, qos: Int = 0) { + synchronized(subscriptions) { subscriptions[topic] = qos } + scope.launch { + // 未连接时仅记录订阅关系，待连接成功后由 connectComplete 恢复， + // 不发 Subscribed 事件以免误报"订阅成功" + val current = client ?: run { + Timber.d("MQTT 未连接，订阅 $topic 已记录，连接成功后自动恢复") + return@launch + } + try { + // subscribe 返回 Broker 实际授予的 QoS（0x80 表示被拒绝），用于联调排查 + current.subscribe(topic, qos) + _messages.emit(MqttMessage.Subscribed(topic)) + Timber.i("MQTT 订阅成功: $topic qos=$qos") + } catch (e: Exception) { + _messages.emit(MqttMessage.Error("订阅失败: ${e.message}", e)) + } + } + } + + fun publish(topic: String, payload: ByteArray, qos: Int = 0, retain: Boolean = false) { + scope.launch { + try { + client?.publish(topic, payload, qos, retain) + _messages.emit(MqttMessage.Published(topic)) + } catch (e: Exception) { + _messages.emit(MqttMessage.Error("发布失败: ${e.message}", e)) + } + } + } + + fun destroy() { + disconnect() + scope.cancel() + } + + /** 使用 Extended 回调：connectComplete 在首次连接与自动重连成功时都会回调，用于恢复订阅 */ + private fun createCallback() = object : MqttCallbackExtended { + override fun connectComplete(reconnect: Boolean, serverURI: String?) { + scope.launch { + _state.value = MqttState.Connected + Timber.i("MQTT connectComplete: reconnect=$reconnect serverURI=$serverURI") + // 重连后恢复历史订阅（cleanSession=false 时 Broker 已保留订阅，重复订阅幂等） + val restore = synchronized(subscriptions) { subscriptions.toMap() } + if (restore.isNotEmpty()) { + try { + client?.subscribe(restore.keys.toTypedArray(), restore.values.toIntArray()) + Timber.i("MQTT ${if (reconnect) "重连" else "连接"}完成，已恢复订阅 ${restore.size} 个: ${restore.keys}") + } catch (e: Exception) { + Timber.e(e, "MQTT 恢复订阅失败") + } + } else { + Timber.w("MQTT 连接完成但无历史订阅（尚未调用过 subscribe）") + } + _messages.emit(MqttMessage.Connected) + } + } + + override fun connectionLost(cause: Throwable?) { + scope.launch { + _state.value = MqttState.Disconnected + _messages.emit(MqttMessage.Disconnected(cause)) + // cause 可能为 null（如 Broker 主动踢 clientId 重复时仅表现为静默断开）， + // 打印消息体帮助定位掉线原因 + Timber.w(cause, "MQTT 连接丢失: ${cause?.message ?: "无异常信息（可能是 Broker 踢线/网络中断）"}") + } + } + + override fun messageArrived(topic: String?, message: org.eclipse.paho.client.mqttv3.MqttMessage?) { + if (topic != null && message != null) { + Timber.i( + "MQTT messageArrived: topic=$topic size=${message.payload.size} " + + "qos=${message.qos} dup=${message.isDuplicate} retained=${message.isRetained}" + ) + scope.launch { + _messages.emit(MqttMessage.Received(topic, message.payload)) + } + } else { + Timber.w("MQTT messageArrived 收到空消息: topic=$topic message=$message") + } + } + + override fun deliveryComplete(token: IMqttDeliveryToken?) {} + } +} diff --git a/app/src/main/java/com/sw/platecabinet/mqtt/MqttMessage.kt b/app/src/main/java/com/sw/platecabinet/mqtt/MqttMessage.kt new file mode 100644 index 0000000..9a04d0d --- /dev/null +++ b/app/src/main/java/com/sw/platecabinet/mqtt/MqttMessage.kt @@ -0,0 +1,11 @@ +package com.sw.platecabinet.mqtt + +/** MQTT 事件消息密封类 */ +sealed class MqttMessage { + data object Connected : MqttMessage() + data class Disconnected(val cause: Throwable? = null) : MqttMessage() + data class Received(val topic: String, val payload: ByteArray) : MqttMessage() + data class Subscribed(val topic: String) : MqttMessage() + data class Published(val topic: String) : MqttMessage() + data class Error(val message: String, val throwable: Throwable? = null) : MqttMessage() +} diff --git a/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt b/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt index cb6dc1d..4e3fea4 100644 --- a/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt +++ b/app/src/main/java/com/sw/platecabinet/viewmodel/NetViewModelV2.kt @@ -137,7 +137,7 @@ class NetViewModelV2 : ViewModel() { } ErrorInfo.MERR_ASF_ALREADY_ACTIVATED -> { -// ToastUtils.showToast("引擎已激活，无需再次激活") + ToastUtils.showToast("引擎已激活，无需再次激活") } else -> { diff --git a/app/src/main/res/layout/activity_init.xml b/app/src/main/res/layout/activity_init.xml index 0c5aa80..f50c1dd 100644 --- a/app/src/main/res/layout/activity_init.xml +++ b/app/src/main/res/layout/activity_init.xml @@ -12,52 +12,40 @@ android:id="@+id/includeHeader" layout="@layout/item_title_time" /> - - + + android:adjustViewBounds="true" + android:importantForAccessibility="no" + android:src="@drawable/ic_init_text" /> + android:adjustViewBounds="true" + android:importantForAccessibility="no" + android:src="@drawable/ic_init_img" /> - - - - - - - + android:gravity="center" + android:paddingStart="45dp" + android:paddingEnd="45dp" + android:text="@string/init_take_plate" + android:textColor="@color/init_button_text" + android:textSize="24sp" /> - - - \ No newline at end of file + diff --git a/app/src/main/res/layout/dialog_environment_select.xml b/app/src/main/res/layout/dialog_environment_select.xml new file mode 100644 index 0000000..c08c614 --- /dev/null +++ b/app/src/main/res/layout/dialog_environment_select.xml @@ -0,0 +1,36 @@ + + + + + + + + + diff --git a/app/src/main/res/values/colors.xml b/app/src/main/res/values/colors.xml index 1f5285e..8cf6193 100644 --- a/app/src/main/res/values/colors.xml +++ b/app/src/main/res/values/colors.xml @@ -15,4 +15,5 @@ #ff02f1be #FFCC99 #FFCC99 + #FFCC99 \ No newline at end of file diff --git a/app/src/main/res/values/strings.xml b/app/src/main/res/values/strings.xml index ed80ad9..e955d7b 100644 --- a/app/src/main/res/values/strings.xml +++ b/app/src/main/res/values/strings.xml @@ -1,5 +1,6 @@ 餐盘柜会员版 + 点击取盘 Hello blank fragment diff --git a/gradle/libs.versions.toml b/gradle/libs.versions.toml index 90ead1d..b33c6a3 100644 --- a/gradle/libs.versions.toml +++ b/gradle/libs.versions.toml @@ -22,6 +22,7 @@ okhttp = "4.12.0" timber = "5.0.1" gson = "2.13.1" hiltAndroid = "2.56.2" +paho = "1.2.5" core = "3.4.1" androidCore = "3.3.0" @@ -58,6 +59,7 @@ okhttp = { module = "com.squareup.okhttp3:okhttp", version.ref = "okhttp" } retrofit = { module = "com.squareup.retrofit2:retrofit", version.ref = "retrofit" } converter-gson = { module = "com.squareup.retrofit2:converter-gson", version.ref = "retrofit" } timber = { module = "com.jakewharton.timber:timber", version.ref = "timber" } +paho-mqtt = { group = "org.eclipse.paho", name = "org.eclipse.paho.client.mqttv3", version.ref = "paho" } [plugins] android-application = { id = "com.android.application", version.ref = "agp" } diff --git a/lib_face/src/main/java/com/sw/plate/utils/arcface/FaceRectTransformer.java b/lib_face/src/main/java/com/sw/plate/utils/arcface/FaceRectTransformer.java index 58c9d5d..9db78ee 100644 --- a/lib_face/src/main/java/com/sw/plate/utils/arcface/FaceRectTransformer.java +++ b/lib_face/src/main/java/com/sw/plate/utils/arcface/FaceRectTransformer.java @@ -77,7 +77,6 @@ public class FaceRectTransformer { rect.bottom *= verticalRatio; Rect newRect = new Rect(); - L.e("cameraDisplayOrientation " + cameraDisplayOrientation + " === " + cameraId); switch (cameraDisplayOrientation) { case 0: if (cameraId == Camera.CameraInfo.CAMERA_FACING_FRONT) { diff --git a/lib_face/src/main/java/com/sw/plate/utils/arcface/face/FaceHelper.java b/lib_face/src/main/java/com/sw/plate/utils/arcface/face/FaceHelper.java index b7a11b1..6f333d4 100644 --- a/lib_face/src/main/java/com/sw/plate/utils/arcface/face/FaceHelper.java +++ b/lib_face/src/main/java/com/sw/plate/utils/arcface/face/FaceHelper.java @@ -246,7 +246,6 @@ public class FaceHelper implements FaceListener { * @param format 图像格式 */ public void requestFaceFeature(byte[] nv21, FacePreviewInfo facePreviewInfo, int width, int height, int format) { - L.e("requestFaceFeature===frThreadQueue.remainingCapacity()=" + frThreadQueue.remainingCapacity()); if (frEngine != null && frThreadQueue.remainingCapacity() > 0) { frExecutor.execute(new FaceRecognizeRunnable(nv21, facePreviewInfo, width, height, format)); } else {