biz_xfj_cloud/lib/common/utils/app/app_mq_tool.dart

import 'dart:convert';
import 'dart:async';

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 isNetwork = AppNetworkTool.isAutoNetwork();
    if (!isNetwork) {
      logger.e("|$_tag|mqtt监听消息错误没有网络");
      return;
    }
    final app = Get.find<AppControl>();

    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|");
      }
    });
  }

  /// 根据协议动作分发已完成设备 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;
    switch (msg.action!) {
      case MqtActionEnum.CONFIG:
        await SyncConfigTool.autoSyncSysInfo();
        mqUpdateEventBus.fire(MqtActionEnum.CONFIG);
        break;
      case MqtActionEnum.CANTEEN_PRODUCT:
        await SyncProductTool.syncProduct();
        mqUpdateEventBus.fire(MqtActionEnum.CANTEEN_PRODUCT);
        break;
      case MqtActionEnum.SUPERMARKET_PRODUCT:
        await SyncProductTool.syncProduct();
        mqUpdateEventBus.fire(MqtActionEnum.SUPERMARKET_PRODUCT);
        break;
      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) &&
            ObjectUtil.isNotEmpty(msg.androidOrderCode) &&
            ObjectUtil.isNotEmpty(msg.pic)) {
          SyncPrintTool.addTask(
            orderId: msg.orderId!,
            androidOrderId: msg.androidOrderCode!,
            historyId: msg.historyId!,
            pic: msg.pic!,
          );
        }
        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:
        return;
      // final msg = MqttInvoiceMsg.fromJson(json.decode(orgStr));
      // final storagePath = await AppFileTool.getFacePath();
      // if (ObjectUtil.isEmpty(storagePath)) {
      //   logger.e("|$_tag|mq同步人脸错误,创建人脸存储路径错误");
      //   return;
      // }
      // final sysSetting = await AppSettingTool.getSysSetting();
      // var merchantDeptId = sysSetting.merchantDeptId;
      // if (ObjectUtil.isEmpty(storagePath)) {
      //   logger.e("|$_tag|mq同步人脸错误,merchantId为空");
      //   return;
      // }
      // await SyncFaceTool.syncGroupFace(
      //   pageNo: 1,
      //   userId: msg.userId,
      //   merchantDeptId: merchantDeptId!,
      //   storagePath: storagePath!,
      //   isGlobalFace: msg.isGlobalFace,
      // );
      // mqUpdateEventBus.fire(MqtActionEnum.FACE);
      // break;
    }
  }

  /// 取消当前 MQTT 消息订阅并清空 AppControl 中的监听句柄。
  static void mqListenClose() {
    final app = Get.find<AppControl>();
    app.mqListen?.cancel();
    app.mqListen = null;
  }
}