biz_xfj_cloud/lib/common/utils/manage/mq_manager.dart

import 'dart:io';

import 'package:common_utils/common_utils.dart';
import 'package:flutter/services.dart';
import 'package:mqtt_client/mqtt_client.dart';
import 'package:mqtt_client/mqtt_server_client.dart';

import 'package:pj_supermarket_cloud/common/index.dart';

/// mq
class MqManager {
  static const String _tag = "mq管理";

  static MqManager? _instance;

  // mq对象
  static MqttServerClient? mqttClient;
  static MqttConnectionState state = MqttConnectionState.disconnected;
  static Future<void>? _refreshFuture;
  static String? connectionFailureReason;

  // 当前设备的 topic
  static late String mainTopic;

  MqManager._internal();

  factory MqManager() {
    _instance ??= MqManager._internal();
    return _instance!;
  }

  static Future<SysSetting?> validate() async {
    final config = await AppSettingTool.getSysSetting();
    if (ObjectUtil.isEmpty(config)) return null;
    if (ObjectUtil.isEmpty(config.mqUsername)) return null;
    if (ObjectUtil.isEmpty(config.mqPassword)) return null;
    if (ObjectUtil.isEmpty(config.mqClientId)) return null;
    if (ObjectUtil.isEmpty(config.mqSocketAddress)) return null;
    if (ObjectUtil.isEmpty(config.mqSocketPort)) return null;
    if (ObjectUtil.isEmpty(config.mainTopic)) return null;
    // 只记录非敏感连接信息，避免账号、密码等配置进入终端日志。
    logger.i(
      "|$_tag|mq配置校验通过,地址:${config.mqSocketAddress},端口:${config.mqSocketPort}",
    );
    return config;
  }

  static Future<void> closeMqtt() async {
    if (MqManager.mqttClient != null) {
      MqManager.mqttClient?.disconnect();
      MqManager.mqttClient = null;
    }
    MqManager.state = MqttConnectionState.disconnected;
    connectionFailureReason = null;
  }

  static Future<void> initMqtt() async {
    // 获取配置
    final config = await validate();
    if (config == null) return;
    final port = int.tryParse(config.mqSocketPort!);
    if (ObjectUtil.isEmpty(port)) return;
    final client = MqttServerClient.withPort(
      config.mqSocketAddress!,
      config.mqClientId!,
      port!,
    );
    mqttClient = client;

    if (port == 8883) {
      final context = SecurityContext.defaultContext;
      final byte = await rootBundle.load(AssetsCrt.emqxCa);
      context.setClientAuthoritiesBytes(byte.buffer.asUint8List());
      client.securityContext = context;
      client.secure = true;
    }

    client.onBadCertificate = (Object a) => true;
    client.autoReconnect = false;
    client.logging(on: false);
    client.keepAlivePeriod = 60;
    client.onConnected = _onConnected;
    client.onDisconnected = _onDisconnected;
    client.onSubscribed = _onSubscribed;
    client.pongCallback = _pongCallback;
    client.onAutoReconnect = _onAutoReconnect;

    final connMessage =
        MqttConnectMessage().withProtocolName("MQTT").withProtocolVersion(4);
    client.connectionMessage = connMessage;

    try {
      final result = await client.connect(config.mqUsername, config.mqPassword);
      MqManager.state = result?.state ?? MqttConnectionState.disconnected;
      connectionFailureReason =
          MqManager.state == MqttConnectionState.connected ? null : '消息服务未建立连接';
    } catch (e, stack) {
      _recordConnectionFailure(e, stack);
    }
  }

  static Future<void> refresh() async {
    final refreshing = _refreshFuture;
    if (refreshing != null) return refreshing;
    final future = _refresh();
    _refreshFuture = future;
    try {
      await future;
    } finally {
      if (identical(_refreshFuture, future)) _refreshFuture = null;
    }
  }

  /// 网络监听与启动流程可能同时触发重连，串行执行可避免同一客户端并发 connect。
  static Future<void> _refresh() async {
    final config = await validate();
    if (ObjectUtil.isEmpty(config)) {
      logger.e("|$_tag||刷新mq|错误mq参数校验不通过");
      return;
    }
    if (ObjectUtil.isEmpty(mqttClient)) {
      await initMqtt();
    } else if (MqManager.state != MqttConnectionState.connected) {
      try {
        final result = await mqttClient!.connect(
          config!.mqUsername,
          config.mqPassword,
        );
        MqManager.state = result?.state ?? MqttConnectionState.disconnected;
        connectionFailureReason =
            MqManager.state == MqttConnectionState.connected
                ? null
                : '消息服务未建立连接';
      } catch (e, stack) {
        _recordConnectionFailure(e, stack);
      }
    }
  }

  static Future<void> subscribeCurrentTopic() async {
    if (ObjectUtil.isEmpty(mqttClient)) return;
    final config = await validate();
    if (ObjectUtil.isEmpty(config)) return;
    if (ObjectUtil.isEmpty(config?.mainTopic)) return;
    mqttClient!.subscribe(config!.mainTopic!, MqttQos.atMostOnce);
  }

  static void _onConnected() async {
    logger.w("|$_tag|mqtt连接成功");
    connectionFailureReason = null;
    final config = await AppSettingTool.getSysSetting();
    if (ObjectUtil.isNotEmpty(config)) {
      config.mqSyncTime = DateTime.now();
      await AppSettingTool.setSysSetting(config);
    }
    await subscribeCurrentTopic();
  }

  static void _onDisconnected() async {
    logger.w("|$_tag|mqtt断开链接");
    AppMqTool.mqListenClose();
    state = MqttConnectionState.disconnected;
  }

  static void _onSubscribed(String topic) {
    logger.w('|$_tag|mqtt 订阅主题成功:$topic');
    mainTopic = topic;
  }

  static void _pongCallback() {
    logger.w("|$_tag|mqtt ping");
  }

  // 自动重连接
  static void _onAutoReconnect() async {
    logger.i("|$_tag|启动自动重连");
    await refresh();
  }

  static void _recordConnectionFailure(Object error, StackTrace stack) {
    state = MqttConnectionState.disconnected;
    final message = error.toString().toLowerCase();
    connectionFailureReason = message.contains('connection refused')
        ? '消息服务拒绝连接'
        : message.contains('timed out') || message.contains('timeout')
            ? '消息服务连接超时'
            : error is SocketException
                ? '消息服务地址不可达'
                : '消息服务连接失败';
    logger.e("|$_tag|mqtt连接失败error:|$error|stack:|$stack|");
  }
}