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

import 'dart:async';
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 const Duration _connectionTimeout = Duration(seconds: 10);

  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 {
    final client = MqManager.mqttClient;
    if (client != null) _discardClient(client);
    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);
    // mqtt_client 自带参数只约束等待 Broker 确认的间隔，外层 Future 超时还需兜底 DNS/TCP/TLS 卡死。
    client.connectTimeoutPeriod = 3000;
    client.keepAlivePeriod = 60;
    client.onConnected = () => _onConnected(client);
    client.onDisconnected = () => _onDisconnected(client);
    client.onSubscribed = _onSubscribed;
    client.pongCallback = _pongCallback;
    client.onAutoReconnect = () => _onAutoReconnect(client);

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

    await _connect(client, config);
  }

  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) {
      await _connect(mqttClient!, config!);
    }
  }

  static bool get isConnected =>
      state == MqttConnectionState.connected && mqttClient != null;

  /// 所有 MQTT 建连统一从这里进入，保证界面操作和后台重试都有明确的等待上限。
  static Future<void> _connect(
    MqttServerClient client,
    SysSetting config,
  ) async {
    final connectFuture = client.connect(config.mqUsername, config.mqPassword);
    try {
      final result = await connectFuture.timeout(_connectionTimeout);
      if (!identical(mqttClient, client)) {
        _disconnectClient(client);
        return;
      }
      state = result?.state ?? MqttConnectionState.disconnected;
      connectionFailureReason =
          state == MqttConnectionState.connected ? null : '消息服务未建立连接';
      if (state != MqttConnectionState.connected) _discardClient(client);
    } on TimeoutException catch (e, stack) {
      _recordConnectionFailure(e, stack);
      // Future.timeout 不会取消底层 Socket.connect。先摘除全局引用释放重试锁，
      // 保留 client 内部 handler，待原连接结束后才能可靠关闭可能晚到的底层 Socket。
      _detachClient(client);
      unawaited(_disconnectWhenConnectSettles(connectFuture, client));
    } catch (e, stack) {
      _recordConnectionFailure(e, stack);
      _discardClient(client);
    }
  }

  static Future<void> _disconnectWhenConnectSettles(
    Future<MqttClientConnectionStatus?> connectFuture,
    MqttServerClient client,
  ) async {
    try {
      await connectFuture;
    } catch (_) {
      // 原始连接异常已在超时分支记录，此处仅负责释放晚到的底层连接。
    } finally {
      _disconnectClient(client);
    }
  }

  static void _discardClient(MqttServerClient client) {
    _disconnectClient(client);
    _detachClient(client);
  }

  static void _detachClient(MqttServerClient client) {
    if (identical(mqttClient, client)) mqttClient = null;
    state = MqttConnectionState.disconnected;
  }

  static void _disconnectClient(MqttServerClient client) {
    try {
      client.disconnect();
    } catch (e, stack) {
      logger.w('|$_tag|释放mqtt连接失败error:|$e|stack:|$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(MqttServerClient client) async {
    if (!identical(mqttClient, client)) {
      _disconnectClient(client);
      return;
    }
    logger.w("|$_tag|mqtt连接成功");
    state = MqttConnectionState.connected;
    connectionFailureReason = null;
    final config = await AppSettingTool.getSysSetting();
    if (ObjectUtil.isNotEmpty(config)) {
      config.mqSyncTime = DateTime.now();
      await AppSettingTool.setSysSetting(config);
    }
    await subscribeCurrentTopic();
  }

  static void _onDisconnected(MqttServerClient client) async {
    if (!identical(mqttClient, client)) return;
    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(MqttServerClient client) async {
    if (!identical(mqttClient, client)) return;
    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
                ? '消息服务地址不可达'
                : '消息服务连接失败';
    if (error is TimeoutException || error is SocketException) {
      // 连接超时和地址不可达属于可预期的外部依赖故障，任务日志会保留结构化结果，
      // 此处不再每分钟重复输出整段堆栈污染待上传日志。
      logger.w("|$_tag|mqtt连接失败:${connectionFailureReason ?? error}|");
      return;
    }
    logger.e("|$_tag|mqtt连接失败error:|$error|stack:|$stack|");
  }
}