terrylee/TaiwanStar_general_size_s_Chengwai
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix(mqtt/ota): 修同一遠端指令被投遞多次致 OTA 下載互相取消而失敗

Browse Source 
- MqttManager: 收訊改每個 client 註冊一次的全域 publishes 回呼；subscribeToCommandTopic 只送 SUBSCRIBE 不再掛 callback(每次重連累積 callback=元凶)；加 command_id 去重(30s)
- OtaManager: startUpdate 加同 cmdId 防重入 guard，避免並行 cancelExisting 互相取消下載

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
This commit is contained in:
terrylee 2026-07-16 14:07:05 +08:00
parent bfaad4ea3b
commit d224354c23
2 changed files with 57 additions and 12 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

50
app/src/main/java/com/unibuy/smartdevice/external/mqtt/MqttManager.java vendored
View File

@ -27,6 +27,10 @@ public class MqttManager {
private static final String TAG = "MqttManager";
private static MqttManager instance;
private Mqtt3AsyncClient client;
// 指令去重（保險）：同一 command_id 短時間內只處理一次，防 MQTT 重複投遞（QoS1 重送等）導致同指令被執行多次
private final java.util.Map<String, Long> recentCmdIds = new java.util.concurrent.ConcurrentHashMap<>();
private static final long CMD_DEDUP_WINDOW_MS = 30000L;
private final Gson gson = new Gson();
private com.unibuy.smartdevice.exception.Logs logs;
private String serialNo;
@ -129,6 +133,38 @@ public class MqttManager {
})
.buildAsync();
// 收訊回呼「每個 client 只註冊一次」的全域 publishes（取代每次重連 subscribe 都掛一個 callback 的舊寫法）。
// 舊寫法會在一整天斷線重連中累積數十個 callback，導致同一則指令被處理數十次（OTA 因此並行互相取消而失敗）。
final String globalCommandTopic = "machine/" + machineId + "/command";
client.publishes(com.hivemq.client.mqtt.MqttGlobalPublishFilter.ALL, publish -> {
try {
if (!globalCommandTopic.equals(publish.getTopic().toString())) {
return;
}
String payloadString = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
logs.info("Received command payload: " + payloadString);
CommandPayload command = gson.fromJson(payloadString, CommandPayload.class);
if (commandListener != null && command != null) {
// 去重：同一 command_id 短時間內重複投遞只處理一次
String cid = command.getCommand_id();
if (cid != null && !cid.isEmpty()) {
long nowMs = System.currentTimeMillis();
Long lastMs = recentCmdIds.put(cid, nowMs);
if (lastMs != null && (nowMs - lastMs) < CMD_DEDUP_WINDOW_MS) {
logs.info("Duplicate MQTT command ignored (ID: " + cid + ")");
return;
}
if (recentCmdIds.size() > 200) {
recentCmdIds.entrySet().removeIf(en -> nowMs - en.getValue() > 60000L);
}
}
commandListener.onCommandReceived(command);
}
} catch (Exception e) {
logs.info("Error parsing command payload: " + e.getMessage());
}
});
client.connectWith()
.cleanSession(true)
.keepAlive(30) // Broker 判斷斷線 = 30 × 1.5 = 45 秒，加快 LWT 觸發
@ -252,21 +288,11 @@ public class MqttManager {
return;
String commandTopic = "machine/" + serialNo + "/command";
// 只送 SUBSCRIBE 建立訂閱（cleanSession 每次重連都要重送）；不再於此掛 callback，
// 收訊統一走 connect() 內註冊一次的全域 publishes 回呼，避免重連累積多個 callback。
client.subscribeWith()
.topicFilter(commandTopic)
.qos(MqttQos.AT_LEAST_ONCE)
.callback(publish -> {
String payloadString = new String(publish.getPayloadAsBytes(), StandardCharsets.UTF_8);
logs.info("Received command payload: " + payloadString);
try {
CommandPayload command = gson.fromJson(payloadString, CommandPayload.class);
if (commandListener != null && command != null) {
commandListener.onCommandReceived(command);
}
} catch (Exception e) {
logs.info("Error parsing command payload: " + e.getMessage());
}
})
.send()
.whenComplete((subAck, throwable) -> {
if (throwable != null) {

19
app/src/main/java/com/unibuy/smartdevice/tools/OtaManager.java
View File

@ -65,6 +65,13 @@ public class OtaManager {
private static volatile boolean awaitingIdleInstall = false;
private static BroadcastReceiver completionReceiver;
// 防重入：MQTT 可能重複投遞同一則 update_app（機台若有多條堆疊連線會收到 N 次）。
// 同一 cmdId 已在進行中就忽略重複，避免 N 個並行 startUpdate 經 cancelExisting() 互相取消下載。
// 不在 clearState 清除（cancelExisting 會於 startUpdate 開頭呼叫 clearState），改由新 cmdId 覆蓋 + 15 分鐘保險過期。
private static volatile String activeOtaCmdId = null;
private static volatile long activeOtaStartMs = 0L;
private static final long OTA_REENTRY_WINDOW_MS = 15 * 60 * 1000L;
/** 相容舊呼叫端：未帶版本資訊。 */
public static void startUpdate(Context context, String apkUrl, String cmdId) {
startUpdate(context, apkUrl, cmdId, null);
@ -81,6 +88,18 @@ public class OtaManager {
public static void startUpdate(Context context, String apkUrl, String cmdId, String expectedVersion) {
Context appContext = context.getApplicationContext();
MqttManager mqttManager = MqttManager.getInstance();
// 防重入：同一 cmdId 的重複投遞直接忽略，避免並行 startUpdate 互相 cancelExisting() 取消下載。
synchronized (OtaManager.class) {
long now = System.currentTimeMillis();
if (cmdId != null && cmdId.equals(activeOtaCmdId) && (now - activeOtaStartMs) < OTA_REENTRY_WINDOW_MS) {
logs.info("OTA duplicate command ignored (cmdId=" + cmdId + "); already in progress.");
return;
}
activeOtaCmdId = cmdId;
activeOtaStartMs = now;
}
try {
logs.info("Starting OTA update process. URL: " + apkUrl + " (cmdId: " + cmdId
+ ", version: " + expectedVersion + ")");