1优化改动设备栏中全部标签栏的设备列表中数据的整合，又数据合并解耦成使用已有组件各自维护，使其样式和逻辑一致。

2接入了无人机机场的实时信息和无人机的实时信息，并创建了对应的组件，并测试。 3.实现了机器人的详情页的视频显示的逻辑为当机器人在线，就在下方同步出来机器人的各个方位视频，并且支持切换。 4.实现了全部设备中对类型的区分和实现了点击进入对应的详情页。 5.修复了部分bug，优化了代码结构，提高了代码的可读性和可维护性。 6更改了机器人列表项中的信息展示为机器序列号和别名和id（优先展示别名） 7.更改了无人机场的信息展示上位无人机及序列号下位机场序列号。 8优化了无人机机场信息只能是的为明文展示。便于观看理解。
2026-06-23 16:20:51 +08:00
parent 06718ed957
commit b7ea788298
25 changed files with 3074 additions and 309 deletions
--- a/lib/core/di/injection.dart
+++ b/lib/core/di/injection.dart
@@ -192,11 +192,14 @@ Future<void> init() async {
  sl.registerLazySingleton<ILoggerService>(() => SentryLoggerImpl());

  /// 1.4 --- MQTT Data Sources ---
-  sl.registerLazySingleton<DroneOsdDataSource>(
-    () => DroneOsdDataSourceImpl(sl<MqttClient>(instanceName: 'droneOsdClient')),
+  sl.registerFactory<DroneOsdDataSource>(
+    () =>
+        DroneOsdDataSourceImpl(sl<MqttClient>(instanceName: 'droneOsdClient')),
  );
-  sl.registerLazySingleton<TaskMessageDataSource>(
-    () => TaskMessageDataSourceImpl(sl<MqttClient>(instanceName: 'taskMessageClient')),
+  sl.registerFactory<TaskMessageDataSource>(
+    () => TaskMessageDataSourceImpl(
+      sl<MqttClient>(instanceName: 'taskMessageClient'),
+    ),
  );

  /// 1.5 --- MQTT Repositories ---
@@ -492,9 +495,7 @@ Future<void> init() async {
  );

  /// 创建设备任务（通过接口执行作业）
-  sl.registerLazySingleton(
-    () => CreateDeviceTaskUseCase(sl<PathRepository>()),
-  );
+  sl.registerLazySingleton(() => CreateDeviceTaskUseCase(sl<PathRepository>()));

  /// 6. 认证 (Auth)
  // --- 关键修改点 1: AuthCubit 必须在 GoRouter 之前注册，并传入参数 ---
@@ -546,12 +547,5 @@ Future<void> init() async {
  sl.registerLazySingleton(() => CancelTaskUseCase(sl()));
  sl.registerLazySingleton(() => PauseTaskUseCase(sl()));
  sl.registerLazySingleton(() => RecoveryTaskUseCase(sl()));
-  sl.registerFactory(
-    () => DeviceTaskCubit(
-      sl(),
-      sl(),
-      sl(),
-      sl(),
-    ),
-  );
+  sl.registerFactory(() => DeviceTaskCubit(sl(), sl(), sl(), sl()));
 }
--- a/lib/core/network/mqtt/data/datasources/drone_osd_datasource.dart
+++ b/lib/core/network/mqtt/data/datasources/drone_osd_datasource.dart
@@ -40,6 +40,17 @@ class DroneOsdDataSourceImpl implements DroneOsdDataSource {
    required String deviceSn,
    required String gatewaySn,
  }) async {
+    if (_deviceSn == deviceSn &&
+        _gatewaySn == gatewaySn &&
+        _subscription != null) {
+      debugPrint('[DroneOsdDataSource] already listening');
+      return;
+    }
+
+    if (_deviceSn != null || _gatewaySn != null) {
+      await stopListening();
+    }
+
    _deviceSn = deviceSn;
    _gatewaySn = gatewaySn;

@@ -50,8 +61,12 @@ class DroneOsdDataSourceImpl implements DroneOsdDataSource {
    debugPrint('   无人机: $droneTopic');
    debugPrint('   机场: $stationTopic');

-    await mqttClient.subscribe(droneTopic);
-    await mqttClient.subscribe(stationTopic);
+    if (deviceSn.isNotEmpty) {
+      await mqttClient.subscribe(droneTopic);
+    }
+    if (gatewaySn.isNotEmpty) {
+      await mqttClient.subscribe(stationTopic);
+    }

    _subscription = mqttClient.messageStream?.listen((message) {
      _handleMessage(message);
@@ -63,11 +78,13 @@ class DroneOsdDataSourceImpl implements DroneOsdDataSource {
    await _subscription?.cancel();
    _subscription = null;

-    if (_deviceSn != null) {
-      await mqttClient.unsubscribe('thing/product/$_deviceSn/osd');
+    if (_deviceSn != null && _deviceSn!.isNotEmpty) {
+      final deviceSn = _deviceSn!;
+      await mqttClient.unsubscribe('thing/product/$deviceSn/osd');
    }
-    if (_gatewaySn != null) {
-      await mqttClient.unsubscribe('thing/product/$_gatewaySn/osd');
+    if (_gatewaySn != null && _gatewaySn!.isNotEmpty) {
+      final gatewaySn = _gatewaySn!;
+      await mqttClient.unsubscribe('thing/product/$gatewaySn/osd');
    }

    _deviceSn = null;
@@ -79,12 +96,22 @@ class DroneOsdDataSourceImpl implements DroneOsdDataSource {
      final jsonData = jsonDecode(message.payload) as Map<String, dynamic>;
      final osdData = DroneOsdEntity.fromJson(jsonData);

-      if (message.topic.contains(_deviceSn ?? '')) {
-        debugPrint('🛸 [DroneOsdDataSource] 无人机 OSD 更新');
-        _droneOsdController.add(osdData);
-      } else if (message.topic.contains(_gatewaySn ?? '')) {
+      // 🔥 重要修复：先判断 gatewaySn，再判断 deviceSn
+      // 因为 topic 可能同时包含两者，但我们需要优先匹配机场
+      if (_gatewaySn != null &&
+          _gatewaySn!.isNotEmpty &&
+          message.topic.contains(_gatewaySn!)) {
        debugPrint('🏢 [DroneOsdDataSource] 机场 OSD 更新');
+        debugPrint('🏢 [DroneOsdDataSource] Topic: ${message.topic}');
        _stationOsdController.add(osdData);
+      } else if (_deviceSn != null &&
+          _deviceSn!.isNotEmpty &&
+          message.topic.contains(_deviceSn!)) {
+        debugPrint('🛸 [DroneOsdDataSource] 无人机 OSD 更新');
+        debugPrint('🛸 [DroneOsdDataSource] Topic: ${message.topic}');
+        _droneOsdController.add(osdData);
+      } else {
+        debugPrint('⚠️ [DroneOsdDataSource] 未知设备类型，Topic: ${message.topic}');
      }
    } catch (e) {
      debugPrint('❌ [DroneOsdDataSource] 解析 OSD 数据失败: $e');
--- a/lib/core/network/mqtt/data/datasources/task_message_datasource.dart
+++ b/lib/core/network/mqtt/data/datasources/task_message_datasource.dart
@@ -21,7 +21,8 @@ class TaskMessageDataSourceImpl implements TaskMessageDataSource {
  final MqttClient mqttClient;
  final _taskStatusController = StreamController<TaskStatusEntity>.broadcast();
  final _taskArriveController = StreamController<TaskArriveEntity>.broadcast();
-  final _realTimeMessageController = StreamController<RealTimeMessageEntity>.broadcast();
+  final _realTimeMessageController =
+      StreamController<RealTimeMessageEntity>.broadcast();

  StreamSubscription<MqttMessage>? _subscription;
  String? _deviceId;
@@ -40,6 +41,18 @@ class TaskMessageDataSourceImpl implements TaskMessageDataSource {

  @override
  Future<void> startListening({required String deviceId}) async {
+    if (_deviceId == deviceId && _subscription != null) {
+      debugPrint('[TaskMessageDataSource] already listening: $deviceId');
+      return;
+    }
+
+    if (_deviceId != null && _deviceId != deviceId) {
+      await stopListening();
+    } else {
+      await _subscription?.cancel();
+      _subscription = null;
+    }
+
    _deviceId = deviceId;

    final taskStatusTopic = 'task/$deviceId/status';
@@ -66,9 +79,10 @@ class TaskMessageDataSourceImpl implements TaskMessageDataSource {
    _subscription = null;

    if (_deviceId != null) {
-      await mqttClient.unsubscribe('task/$_deviceId/status');
-      await mqttClient.unsubscribe('task/$_deviceId/arrive');
-      await mqttClient.unsubscribe('device/$_deviceId/realTimeMessage');
+      final deviceId = _deviceId!;
+      await mqttClient.unsubscribe('task/$deviceId/status');
+      await mqttClient.unsubscribe('task/$deviceId/arrive');
+      await mqttClient.unsubscribe('device/$deviceId/realTimeMessage');
    }

    _deviceId = null;
@@ -88,7 +102,9 @@ class TaskMessageDataSourceImpl implements TaskMessageDataSource {
        _taskArriveController.add(arriveInfo);
      } else if (message.topic.contains('/realTimeMessage')) {
        final realTimeMsg = RealTimeMessageEntity.fromJson(jsonData);
-        debugPrint('💬 [TaskMessageDataSource] 实时消息 - 类型: ${realTimeMsg.type}, 数据点数: ${realTimeMsg.data.length}');
+        debugPrint(
+          '💬 [TaskMessageDataSource] 实时消息 - 类型: ${realTimeMsg.type}, 数据点数: ${realTimeMsg.data.length}',
+        );
        _realTimeMessageController.add(realTimeMsg);
      }
    } catch (e) {
--- a/lib/core/network/mqtt/data/infrastructure/mqtt_client_impl.dart
+++ b/lib/core/network/mqtt/data/infrastructure/mqtt_client_impl.dart
@@ -1,4 +1,5 @@
 import 'dart:async';
+
 import 'package:flutter/foundation.dart';
 import 'package:mqtt_client/mqtt_client.dart' as mqtt;
 import 'package:mqtt_client/mqtt_server_client.dart' as mqtt_server;
@@ -10,8 +11,14 @@ import '../../domain/models/mqtt_message.dart';
 class MqttClientImpl implements MqttClient {
  mqtt_server.MqttServerClient? _client;
  final _messageController = StreamController<MqttMessage>.broadcast();
+  final Map<String, int> _subscriptionRefs = <String, int>{};
+
  MqttConfig? _currentConfig;
+  StreamSubscription? _updatesSubscription;
+  Timer? _reconnectTimer;
  bool _isConnected = false;
+  bool _isConnecting = false;
+  bool _manualDisconnect = false;

  @override
  Stream<MqttMessage>? get messageStream => _messageController.stream;
@@ -21,148 +28,239 @@ class MqttClientImpl implements MqttClient {

  @override
  Future<void> connect(MqttConfig config) async {
+    if (_isConnecting) {
+      debugPrint('[MqttClient] connect ignored, already connecting');
+      return;
+    }
+
    if (_isConnected) {
-      debugPrint('⚠️ [MqttClient] 已连接，先断开');
+      debugPrint('[MqttClient] already connected, disconnect first');
      await disconnect();
    }

+    _manualDisconnect = false;
+    _currentConfig = config;
+    _reconnectTimer?.cancel();
+    _reconnectTimer = null;
+    _isConnecting = true;
+
    try {
-      _currentConfig = config;
-      final clientId = '${config.clientId}_${DateTime.now().millisecondsSinceEpoch}';
-
-      switch (config.protocol) {
-        case MqttProtocol.websocket:
-          _client = mqtt_server.MqttServerClient.withPort(
-            config.host,
-            clientId,
-            config.port,
-          );
-          _client!.useWebSocket = true;
-          break;
-        case MqttProtocol.wss:
-          _client = mqtt_server.MqttServerClient.withPort(
-            config.host,
-            clientId,
-            config.port,
-          );
-          _client!.useWebSocket = true;
-          _client!.secure = true;
-          break;
-        case MqttProtocol.tcp:
-        default:
-          _client = mqtt_server.MqttServerClient.withPort(
-            config.host,
-            clientId,
-            config.port,
-          );
-          break;
-      }
-
-      _client!.logging(on: false);
-      _client!.keepAlivePeriod = config.keepAlivePeriod;
-      _client!.autoReconnect = true;
-      _client!.resubscribeOnAutoReconnect = true;
-      _client!.onDisconnected = _onDisconnected;
-      _client!.onConnected = _onConnected;
-      _client!.onSubscribed = _onSubscribed;
-
-      final connMessage = mqtt.MqttConnectMessage()
-          .withClientIdentifier(clientId)
-          .startClean()
-          .withWillQos(mqtt.MqttQos.atLeastOnce);
-
-      _client!.connectionMessage = connMessage;
-
-      debugPrint('🔌 [MqttClient] 开始连接到 ${config.host}:${config.port}');
+      await _updatesSubscription?.cancel();
+      _updatesSubscription = null;
+      _client = _createClient(config);
+      _configureClient(_client!, config);

+      debugPrint('[MqttClient] connecting to ${config.connectionAddress}');
      await _client!.connect(config.username, config.password);

-      if (_client!.connectionStatus?.state == mqtt.MqttConnectionState.connected) {
+      if (_client!.connectionStatus?.state ==
+          mqtt.MqttConnectionState.connected) {
        _isConnected = true;
-        debugPrint('✅ [MqttClient] 连接成功');
+        debugPrint('[MqttClient] connected');
        _listenToMessages();
+        _resubscribeAll();
      } else {
-        throw Exception('连接失败: ${_client!.connectionStatus?.returnCode}');
+        throw Exception(
+          'MQTT connect failed: ${_client!.connectionStatus?.returnCode}',
+        );
      }
    } catch (e) {
-      debugPrint('❌ [MqttClient] 连接异常: $e');
+      debugPrint('[MqttClient] connect error: $e');
      _isConnected = false;
+      _isConnecting = false;
+      _scheduleReconnect();
      rethrow;
+    } finally {
+      _isConnecting = false;
    }
  }

+  mqtt_server.MqttServerClient _createClient(MqttConfig config) {
+    final clientId =
+        '${config.clientId}_${DateTime.now().millisecondsSinceEpoch}';
+
+    switch (config.protocol) {
+      case MqttProtocol.websocket:
+      case MqttProtocol.wss:
+        final client = mqtt_server.MqttServerClient.withPort(
+          config.connectionAddress,
+          clientId,
+          config.port,
+        );
+        client.useWebSocket = true;
+        client.websocketProtocols = ['mqtt'];
+        return client;
+      case MqttProtocol.tcp:
+        return mqtt_server.MqttServerClient.withPort(
+          config.host,
+          clientId,
+          config.port,
+        );
+    }
+  }
+
+  void _configureClient(
+    mqtt_server.MqttServerClient client,
+    MqttConfig config,
+  ) {
+    client.logging(on: false);
+    client.keepAlivePeriod = config.keepAlivePeriod;
+    client.autoReconnect = false;
+    client.resubscribeOnAutoReconnect = false;
+    client.onDisconnected = _onDisconnected;
+    client.onConnected = _onConnected;
+    client.onSubscribed = _onSubscribed;
+
+    var connMessage = mqtt.MqttConnectMessage()
+        .withClientIdentifier(client.clientIdentifier)
+        .withWillQos(mqtt.MqttQos.atLeastOnce);
+
+    if (config.cleanSession) {
+      connMessage = connMessage.startClean();
+    }
+
+    client.connectionMessage = connMessage;
+  }
+
  @override
  Future<void> disconnect() async {
-    debugPrint('🔌 [MqttClient] 断开连接');
+    debugPrint('[MqttClient] disconnect');
+    _manualDisconnect = true;
+    _reconnectTimer?.cancel();
+    _reconnectTimer = null;
+    await _updatesSubscription?.cancel();
+    _updatesSubscription = null;
    _client?.disconnect();
+    _client = null;
    _isConnected = false;
+    _isConnecting = false;
    _currentConfig = null;
  }

  @override
  Future<void> subscribe(String topic) async {
+    final previousRefs = _subscriptionRefs[topic] ?? 0;
+    _subscriptionRefs[topic] = previousRefs + 1;
+
    if (!_isConnected || _client == null) {
-      throw Exception('MQTT 未连接');
+      debugPrint('[MqttClient] queued subscription while disconnected: $topic');
+      _scheduleReconnect();
+      return;
    }

-    debugPrint('📡 [MqttClient] 订阅主题: $topic');
+    if (previousRefs > 0) {
+      debugPrint('[MqttClient] subscription already active: $topic');
+      return;
+    }
+
+    debugPrint('[MqttClient] subscribe: $topic');
    _client!.subscribe(topic, mqtt.MqttQos.atLeastOnce);
  }

  @override
  Future<void> unsubscribe(String topic) async {
-    if (!_isConnected || _client == null) {
-      throw Exception('MQTT 未连接');
+    final currentRefs = _subscriptionRefs[topic] ?? 0;
+    if (currentRefs <= 1) {
+      _subscriptionRefs.remove(topic);
+    } else {
+      _subscriptionRefs[topic] = currentRefs - 1;
+      debugPrint('[MqttClient] subscription still in use: $topic');
+      return;
    }

-    debugPrint('🔕 [MqttClient] 取消订阅: $topic');
+    if (!_isConnected || _client == null) {
+      debugPrint('[MqttClient] removed queued subscription: $topic');
+      return;
+    }
+
+    debugPrint('[MqttClient] unsubscribe: $topic');
    _client!.unsubscribe(topic);
  }

  @override
  Future<void> publish(String topic, String message) async {
    if (!_isConnected || _client == null) {
-      throw Exception('MQTT 未连接');
+      throw Exception('MQTT is not connected');
    }

-    debugPrint('📤 [MqttClient] 发布消息到 $topic');
+    debugPrint('[MqttClient] publish: $topic');
    final builder = mqtt.MqttClientPayloadBuilder();
    builder.addString(message);
    _client!.publishMessage(topic, mqtt.MqttQos.atLeastOnce, builder.payload!);
  }

  void _listenToMessages() {
-    _client!.updates!.listen((List<mqtt.MqttReceivedMessage<mqtt.MqttMessage>> messages) {
-      for (final msg in messages) {
-        final topic = msg.topic;
-        final payload = mqtt.MqttPublishPayload.bytesToStringAsString((msg.payload as mqtt.MqttPublishMessage).payload.message);
+    _updatesSubscription = _client!.updates!.listen(
+      (List<mqtt.MqttReceivedMessage<mqtt.MqttMessage>> messages) {
+        for (final msg in messages) {
+          final topic = msg.topic;
+          final payload = mqtt.MqttPublishPayload.bytesToStringAsString(
+            (msg.payload as mqtt.MqttPublishMessage).payload.message,
+          );

-        debugPrint('📥 [MqttClient] 收到消息 [$topic]: $payload');
+          debugPrint('[MqttClient] received [$topic]: $payload');
+          _messageController.add(MqttMessage(topic: topic, payload: payload));
+        }
+      },
+      onError: (Object error) {
+        debugPrint('[MqttClient] updates stream error: $error');
+        _isConnected = false;
+        _scheduleReconnect();
+      },
+      cancelOnError: false,
+    );
+  }

-        _messageController.add(MqttMessage(
-          topic: topic,
-          payload: payload,
-        ));
-      }
-    });
+  void _resubscribeAll() {
+    if (_subscriptionRefs.isEmpty || _client == null) return;
+
+    for (final topic in _subscriptionRefs.keys) {
+      debugPrint('[MqttClient] resubscribe: $topic');
+      _client!.subscribe(topic, mqtt.MqttQos.atLeastOnce);
+    }
  }

  void _onConnected() {
-    debugPrint('✅ [MqttClient] 已连接');
+    debugPrint('[MqttClient] connected callback');
    _isConnected = true;
  }

  void _onDisconnected() {
-    debugPrint('❌ [MqttClient] 已断开');
+    debugPrint('[MqttClient] disconnected callback');
    _isConnected = false;
+    if (!_manualDisconnect) {
+      _scheduleReconnect();
+    }
  }

  void _onSubscribed(String topic) {
-    debugPrint('✅ [MqttClient] 订阅成功: $topic');
+    debugPrint('[MqttClient] subscribed: $topic');
+  }
+
+  void _scheduleReconnect() {
+    final config = _currentConfig;
+    if (_manualDisconnect || config == null || _isConnected || _isConnecting) {
+      return;
+    }
+
+    if (_reconnectTimer?.isActive ?? false) return;
+
+    final delay = Duration(milliseconds: config.reconnectDelayMs);
+    debugPrint('[MqttClient] reconnect scheduled in ${delay.inMilliseconds}ms');
+    _reconnectTimer = Timer(delay, () async {
+      if (_manualDisconnect || _isConnected || _isConnecting) return;
+
+      try {
+        debugPrint('[MqttClient] reconnecting...');
+        await connect(config);
+      } catch (e) {
+        debugPrint('[MqttClient] reconnect failed: $e');
+      }
+    });
  }

  void dispose() {
-    _messageController.close();
    disconnect();
+    _messageController.close();
  }
 }
--- a/lib/core/network/mqtt/domain/models/mqtt_config.dart
+++ b/lib/core/network/mqtt/domain/models/mqtt_config.dart
@@ -1,6 +1,16 @@
- import 'package:equatable/equatable.dart';
+import 'package:equatable/equatable.dart';

-enum MqttProtocol { tcp, websocket, wss }
+/// MQTT 传输协议类型
+enum MqttProtocol {
+  /// 纯 TCP 连接（原生 MQTT）
+  tcp,
+  
+  /// WebSocket 连接（MQTT over WebSocket）
+  websocket,
+  
+  /// 加密的 WebSocket 连接（MQTT over WSS）
+  wss,
+}

 class MqttConfig extends Equatable {
  final String host;
@@ -12,6 +22,9 @@ class MqttConfig extends Equatable {
  final bool cleanSession;
  final int keepAlivePeriod;
  final int reconnectDelayMs;
+  
+  /// WebSocket 路径（仅 WebSocket 模式使用）
+  final String? wsPath;

  const MqttConfig({
    required this.host,
@@ -23,8 +36,23 @@ class MqttConfig extends Equatable {
    this.cleanSession = true,
    this.keepAlivePeriod = 60,
    this.reconnectDelayMs = 3000,
+    this.wsPath,
  });

+  /// 获取完整的连接地址
+  String get connectionAddress {
+    switch (protocol) {
+      case MqttProtocol.websocket:
+        return 'ws://$host:$port${wsPath ?? '/mqtt'}';
+      case MqttProtocol.wss:
+        return 'wss://$host:$port${wsPath ?? '/mqtt'}';
+      case MqttProtocol.tcp:
+      default:
+        return host;
+    }
+  }
+
+  /// 无人机/机场 OSD 数据（WebSocket MQTT）
  factory MqttConfig.droneOsd() {
    return const MqttConfig(
      host: '1.95.137.212',
@@ -33,11 +61,13 @@ class MqttConfig extends Equatable {
      password: 'jsmbzn520',
      protocol: MqttProtocol.websocket,
      clientId: 'drone_osd_client',
+      wsPath: '/mqtt',
      cleanSession: true,
      reconnectDelayMs: 3000,
    );
  }

+  /// 任务状态消息（TCP MQTT）
  factory MqttConfig.taskMessage() {
    return const MqttConfig(
      host: '1.95.137.212',
@@ -61,6 +91,7 @@ class MqttConfig extends Equatable {
    bool? cleanSession,
    int? keepAlivePeriod,
    int? reconnectDelayMs,
+    String? wsPath,
  }) {
    return MqttConfig(
      host: host ?? this.host,
@@ -72,6 +103,7 @@ class MqttConfig extends Equatable {
      cleanSession: cleanSession ?? this.cleanSession,
      keepAlivePeriod: keepAlivePeriod ?? this.keepAlivePeriod,
      reconnectDelayMs: reconnectDelayMs ?? this.reconnectDelayMs,
+      wsPath: wsPath ?? this.wsPath,
    );
  }

@@ -86,5 +118,6 @@ class MqttConfig extends Equatable {
        cleanSession,
        keepAlivePeriod,
        reconnectDelayMs,
+        wsPath,
      ];
 }
--- a/lib/core/network/mqtt/mqtt_manager.dart
+++ b/lib/core/network/mqtt/mqtt_manager.dart
@@ -20,20 +20,31 @@ class MqttManager {
      return;
    }

+    debugPrint('🔧 [MqttManager] 开始初始化 MQTT 连接...');
+
+    final droneOsdClient = sl<MqttClient>(instanceName: 'droneOsdClient');
+    final taskMessageClient = sl<MqttClient>(instanceName: 'taskMessageClient');
+
+    // 独立连接两个 MQTT 客户端，互不影响
+    await _connectClient(droneOsdClient, MqttConfig.droneOsd(), 'droneOsdClient');
+    await _connectClient(taskMessageClient, MqttConfig.taskMessage(), 'taskMessageClient');
+
+    _isInitialized = true;
+    debugPrint('✅ [MqttManager] MQTT 初始化完成');
+  }
+
+  Future<void> _connectClient(
+    MqttClient client,
+    MqttConfig config,
+    String clientName,
+  ) async {
    try {
-      debugPrint('🔧 [MqttManager] 开始初始化 MQTT 连接...');
-
-      final droneOsdClient = sl<MqttClient>(instanceName: 'droneOsdClient');
-      final taskMessageClient = sl<MqttClient>(instanceName: 'taskMessageClient');
-
-      await droneOsdClient.connect(MqttConfig.droneOsd());
-      await taskMessageClient.connect(MqttConfig.taskMessage());
-
-      _isInitialized = true;
-      debugPrint('✅ [MqttManager] MQTT 初始化完成');
+      debugPrint('🔌 [MqttManager] 连接 $clientName...');
+      await client.connect(config);
+      debugPrint('✅ [MqttManager] $clientName 连接成功');
    } catch (e) {
-      debugPrint('❌ [MqttManager] MQTT 初始化失败: $e');
-      rethrow;
+      debugPrint('❌ [MqttManager] $clientName 连接失败: $e');
+      // 不抛出异常，允许其他客户端继续连接
    }
  }