209 lines
8.5 KiB
Dart
209 lines
8.5 KiB
Dart
import 'dart:async';
|
|
import 'dart:convert';
|
|
|
|
import 'package:common_utils/common_utils.dart';
|
|
import 'package:get/get.dart';
|
|
import 'package:mqtt_client/mqtt_client.dart';
|
|
|
|
import 'package:pj_supermarket_cloud/common/index.dart';
|
|
|
|
/// 应用级 MQTT 连接初始化、精确 Topic 过滤和业务动作分发工具。
|
|
class AppMqTool {
|
|
/// MQTT 连接与消息分发诊断日志使用的稳定模块标签。
|
|
static const String _tag = "app mq 工具";
|
|
|
|
/// 校验网络和设备绑定后重建 MQTT 连接，并注册当前设备的消息监听。
|
|
static Future<void> mqRefresh() async {
|
|
final app = Get.find<AppControl>();
|
|
if (!app.authenticatedServicesEnabled) {
|
|
logger.w('|$_tag|当前没有有效登录会话，暂不连接消息服务');
|
|
return;
|
|
}
|
|
final isNetwork = AppNetworkTool.isAutoNetwork();
|
|
if (!isNetwork) {
|
|
logger.e("|$_tag|mqtt监听消息错误没有网络");
|
|
return;
|
|
}
|
|
final sysInfo = await AppSettingTool.getSysSetting();
|
|
if (ObjectUtil.isEmpty(sysInfo.serviceId) ||
|
|
ObjectUtil.isEmpty(sysInfo.merchantDeptId) ||
|
|
ObjectUtil.isEmpty(sysInfo.hardGroupId)) {
|
|
logger.w('|$_tag|终端绑定信息不完整，暂不连接消息服务');
|
|
return;
|
|
}
|
|
await MqManager.refresh();
|
|
if (MqManager.state != MqttConnectionState.connected) {
|
|
return;
|
|
}
|
|
if (app.mqListen != null) AppMqTool.mqListenClose();
|
|
final appMerchantDeptId = sysInfo.merchantDeptId!;
|
|
final appHardwareId = sysInfo.serviceId!;
|
|
final appHardGroupId = sysInfo.hardGroupId!;
|
|
app.mqListen = MqManager.mqttClient?.updates?.listen((event) {
|
|
String topic = event[0].topic;
|
|
final msg = event[0].payload as MqttPublishMessage;
|
|
topic = topic.replaceAll('topic/android/', '');
|
|
if (ObjectUtil.isEmpty(topic)) return;
|
|
final splitList = topic.split("/");
|
|
String? merchantDeptId;
|
|
String? hardwareId;
|
|
String? hardGroupId;
|
|
if (splitList.isEmpty) {
|
|
return;
|
|
}
|
|
if (splitList.length == 1) {
|
|
hardGroupId = splitList[0].toString();
|
|
if (appHardGroupId != hardGroupId) return;
|
|
}
|
|
if (splitList.length == 2) {
|
|
hardGroupId = splitList[0].toString();
|
|
merchantDeptId = splitList[1].toString();
|
|
if (appHardGroupId != hardGroupId) return;
|
|
if (appMerchantDeptId != merchantDeptId) return;
|
|
}
|
|
if (splitList.length == 3) {
|
|
hardGroupId = splitList[0].toString();
|
|
merchantDeptId = splitList[1].toString();
|
|
hardwareId = splitList[2].toString();
|
|
if (appHardGroupId != hardGroupId) return;
|
|
if (appMerchantDeptId != merchantDeptId) return;
|
|
if (appHardwareId != hardwareId) return;
|
|
}
|
|
try {
|
|
final message = msg.payload.message;
|
|
String str = MqttPublishPayload.bytesToStringAsString(message);
|
|
logger.i("|$_tag|mqtt 接收到的消息:$str;");
|
|
final mqMsg = MqBaseMsg.fromJson(json.decode(str));
|
|
dispatchMqttMessage(mqMsg, str);
|
|
} catch (e) {
|
|
logger.e("|$_tag|mqtt监听消息错误error:|$e|");
|
|
}
|
|
});
|
|
unawaited(SyncPrintTool.pullPendingTasks());
|
|
}
|
|
|
|
/// 根据协议动作分发已完成设备 Topic 校验的 MQTT 消息。
|
|
///
|
|
/// [orgStr] 仅用于需要专用消息模型的兼容动作；日志拉取采用非阻塞触发，避免占用 MQTT 回调。
|
|
static Future<void> dispatchMqttMessage(MqBaseMsg msg, String orgStr) async {
|
|
if (ObjectUtil.isEmpty(msg.action)) return;
|
|
final app = Get.find<AppControl>();
|
|
final mqUpdateEventBus = app.mqUpdateEventBus;
|
|
var setting = await AppSettingTool.getSysSetting();
|
|
switch (msg.action!) {
|
|
case MqtActionEnum.CONFIG:
|
|
await SyncConfigTool.autoSyncSysInfo();
|
|
try {
|
|
setting = await AppSettingTool.updatePrinterBinding(
|
|
await HardApi.getPrinterBinding(),
|
|
);
|
|
} catch (error, stack) {
|
|
// 打印机绑定是业务终端的附加能力，读取失败不能阻断商品、人脸等基础配置刷新。
|
|
logger.w('|$_tag|刷新打印机绑定失败|error:$error|stack:$stack|');
|
|
}
|
|
mqUpdateEventBus.fire(MqtActionEnum.CONFIG);
|
|
break;
|
|
case MqtActionEnum.CANTEEN_PRODUCT:
|
|
// 食堂变更需要让当前收银页及时刷新，但只查询收到消息时的当天，避免 MQTT 拉取未来一周菜谱。
|
|
final success = await SyncProductTool.syncTodayCanteenProduct();
|
|
if (success) {
|
|
mqUpdateEventBus.fire(MqtActionEnum.CANTEEN_PRODUCT);
|
|
}
|
|
break;
|
|
case MqtActionEnum.SUPERMARKET_PRODUCT:
|
|
// 超市商品变更仍进入现有 productSync 队列，避免 MQTT 回调打断收银操作。
|
|
unawaited(SyncCheckTool.instance.trigger(
|
|
ScheduledTaskType.productSync,
|
|
// MQTT 只负责排队；catchUp 让 idleOnly 任务等待完整 60 秒空闲窗口。
|
|
catchUp: true,
|
|
triggerType: ScheduledTaskTriggerType.mqtt,
|
|
));
|
|
return;
|
|
case MqtActionEnum.SUPERMARKET:
|
|
await SyncProductTool.syncSupermarket();
|
|
mqUpdateEventBus.fire(MqtActionEnum.SUPERMARKET);
|
|
break;
|
|
case MqtActionEnum.CANTEEN_CLASSIFY:
|
|
await SyncProductTool.syncClassify();
|
|
mqUpdateEventBus.fire(MqtActionEnum.CANTEEN_CLASSIFY);
|
|
break;
|
|
case MqtActionEnum.SUPERMARKET_CLASSIFY:
|
|
await SyncProductTool.syncClassify();
|
|
mqUpdateEventBus.fire(MqtActionEnum.SUPERMARKET_CLASSIFY);
|
|
break;
|
|
case MqtActionEnum.PRINT:
|
|
mqUpdateEventBus.fire(MqtActionEnum.PRINT);
|
|
final msg = MqttInvoiceMsg.fromJson(json.decode(orgStr));
|
|
if (ObjectUtil.isNotEmpty(msg.historyId) &&
|
|
ObjectUtil.isNotEmpty(msg.orderId) &&
|
|
(msg.sourceHardId == null ||
|
|
msg.sourceHardId == setting.serviceId) &&
|
|
// targetHardId 是逻辑打印机资产，不是 Android 设备；新协议消息必须匹配本机当前绑定。
|
|
(msg.targetHardId == null ||
|
|
msg.targetHardId == setting.printerHardId) &&
|
|
ObjectUtil.isNotEmpty(msg.pic)) {
|
|
unawaited(SyncPrintTool.addTask(
|
|
orderId: msg.orderId!,
|
|
androidOrderId: msg.androidOrderCode,
|
|
historyId: msg.historyId!,
|
|
pic: msg.pic!,
|
|
printCmd: msg.printCmd,
|
|
printType: msg.printType ?? '1',
|
|
printNo: msg.printNo ?? 1,
|
|
retryVersion: msg.retryVersion ?? 0,
|
|
));
|
|
}
|
|
break;
|
|
case MqtActionEnum.TAKE_FOOD_BROADCAST:
|
|
mqUpdateEventBus.fire(MqtActionEnum.TAKE_FOOD_BROADCAST);
|
|
try {
|
|
final msg = MqttTakeFoodBroadcastMsg.fromJson(json.decode(orgStr));
|
|
SyncTakeFoodBroadcastTool.addTask(msg);
|
|
} catch (e, stack) {
|
|
logger.e("|$_tag|定餐广播执行错误orgStr|$orgStr|error|$e|stack|$stack|");
|
|
}
|
|
return;
|
|
case MqtActionEnum.PULL_TERMINAL_LOG:
|
|
mqUpdateEventBus.fire(MqtActionEnum.PULL_TERMINAL_LOG);
|
|
// MQTT 回调不能等待文件扫描和上传，任务会在支付/识别结束后的安全窗口异步执行。
|
|
unawaited(SyncCheckTool.instance.trigger(
|
|
ScheduledTaskType.terminalLogUpload,
|
|
currentLogOnly: true,
|
|
triggerType: ScheduledTaskTriggerType.mqtt,
|
|
));
|
|
return;
|
|
case MqtActionEnum.FACE:
|
|
try {
|
|
final decoded = json.decode(orgStr);
|
|
if (decoded is! Map) {
|
|
throw const FormatException('FACE 消息不是 JSON 对象');
|
|
}
|
|
final faceMessage = MqttFaceChangeMsg.fromJson(
|
|
Map<String, dynamic>.from(decoded),
|
|
);
|
|
final userId = faceMessage.userId?.trim();
|
|
final parsedUserId = int.tryParse(userId ?? '');
|
|
if (parsedUserId == null || parsedUserId <= 0) {
|
|
throw const FormatException('FACE 消息缺少有效用户标识');
|
|
}
|
|
// MQTT 只作为变更通知；队列按 userId 合并，并由 V3 接口回查服务端最新状态。
|
|
await SyncCheckTool.instance.trigger(
|
|
ScheduledTaskType.departmentFaceSync,
|
|
faceUserIds: {userId!},
|
|
triggerType: ScheduledTaskTriggerType.mqtt,
|
|
);
|
|
} catch (error, stack) {
|
|
logger.w('|$_tag|FACE 消息处理失败|error:$error|stack:$stack|');
|
|
}
|
|
return;
|
|
}
|
|
}
|
|
|
|
/// 取消当前 MQTT 消息订阅并清空 AppControl 中的监听句柄。
|
|
static void mqListenClose() {
|
|
final app = Get.find<AppControl>();
|
|
app.mqListen?.cancel();
|
|
app.mqListen = null;
|
|
}
|
|
}
|