fix:增加talkData的组包逻辑

2024-12-23 15:21:48 +08:00 · 2024-12-23 15:21:48 +08:00 · 6147b86e8b
commit 6147b86e8b
parent 75985e0600
5 changed files with 82 additions and 51 deletions
--- a/lib/main/lockDetail/monitoring/star_chart_h264/star_chart_logic.dart
+++ b/lib/main/lockDetail/monitoring/star_chart_h264/star_chart_logic.dart
@ -102,23 +102,23 @@ class StarChartLogic extends BaseGetXController {
  }

  Future<void> startProcessing() async {
-    state.isButtonDisabled.value = true;
-
-    state.voiceProcessor?.addFrameListener(_onFrame);
-    state.voiceProcessor?.addErrorListener(_onError);
-    try {
-      if (await state.voiceProcessor?.hasRecordAudioPermission() ?? false) {
-        await state.voiceProcessor?.start(state.frameLength, state.sampleRate);
-        state.isProcessing.value =
-            await state.voiceProcessor?.isRecording() ?? false;
-      } else {
-        state.errorMessage.value = 'Recording permission not granted';
-      }
-    } on PlatformException catch (ex) {
-      state.errorMessage.value = 'Failed to start recorder: $ex';
-    } finally {
-      state.isButtonDisabled.value = false;
-    }
+    // state.isButtonDisabled.value = true;
+    //
+    // state.voiceProcessor?.addFrameListener(_onFrame);
+    // state.voiceProcessor?.addErrorListener(_onError);
+    // try {
+    //   if (await state.voiceProcessor?.hasRecordAudioPermission() ?? false) {
+    //     await state.voiceProcessor?.start(state.frameLength, state.sampleRate);
+    //     state.isProcessing.value =
+    //         await state.voiceProcessor?.isRecording() ?? false;
+    //   } else {
+    //     state.errorMessage.value = 'Recording permission not granted';
+    //   }
+    // } on PlatformException catch (ex) {
+    //   state.errorMessage.value = 'Failed to start recorder: $ex';
+    // } finally {
+    //   state.isButtonDisabled.value = false;
+    // }
  }

  void _onError(VoiceProcessorException error) {
@ -127,23 +127,23 @@ class StarChartLogic extends BaseGetXController {

  Future<void> stopProcessing() async {
    // 检查 voiceProcessor 是否已经初始化
-    if (state.voiceProcessor == null) {
-      state.errorMessage.value = 'Voice processor is not initialized.';
-      return;
-    }
-
-    state.isButtonDisabled.value = true;
-    try {
-      await state.voiceProcessor?.stop();
-      state.voiceProcessor?.removeFrameListener(_onFrame);
-      state.udpSendDataFrameNumber = 0;
-    } on PlatformException catch (ex) {
-      state.errorMessage.value = 'Failed to stop recorder: $ex';
-    } finally {
-      state.isProcessing.value =
-          await state.voiceProcessor?.isRecording() ?? false;
-      state.isButtonDisabled.value = false;
-    }
+    // if (state.voiceProcessor == null) {
+    //   state.errorMessage.value = 'Voice processor is not initialized.';
+    //   return;
+    // }
+    //
+    // state.isButtonDisabled.value = true;
+    // try {
+    //   await state.voiceProcessor?.stop();
+    //   state.voiceProcessor?.removeFrameListener(_onFrame);
+    //   state.udpSendDataFrameNumber = 0;
+    // } on PlatformException catch (ex) {
+    //   state.errorMessage.value = 'Failed to stop recorder: $ex';
+    // } finally {
+    //   state.isProcessing.value =
+    //       await state.voiceProcessor?.isRecording() ?? false;
+    //   state.isButtonDisabled.value = false;
+    // }
  }

  int linearToULaw(int pcmVal) {
--- a/lib/talk/startChart/entity/scp_message.dart
+++ b/lib/talk/startChart/entity/scp_message.dart
@ -170,9 +170,12 @@ class ScpMessage {
  static ScpMessage deserialize(List<int> bytes) {
    final message = ScpMessage();
    int offset = 0;
-    // _log(
-    //     text:
-    //         '原始字节数组: ${bytes.sublist(0, 20).map((b) => b.toRadixString(16)).join(" ")}');
+    // Convert byte array to hex string with zero padding and without spaces
+    // String hexString =
+    //     bytes.map((b) => b.toRadixString(16).padLeft(2, '0')).join();
+    //
+    // // Log the hex string
+    // _log(text: '原始字节数组: $hexString');
    // ProtocolFlag (4 bytes)
    if (bytes.length - offset >= 4) {
      message.ProtocolFlag = utf8.decode(bytes.sublist(offset, offset + 4));
@ -257,6 +260,7 @@ class ScpMessage {
    } else {
      throw FormatException("Invalid PayloadLength length");
    }
+    // print("message result: $message");

    // 处理其他类型的Payload
    if (message.PayloadLength != null &&
@ -317,6 +321,7 @@ class ScpMessage {
              spTotal: spTotal,
              spIndex: spIndex,
              byte: byte,
+              payloadType: payloadType,
            );
          } else {
            // 如果 spTotal 为 1 或者没有分包信息，直接处理 byte 数据
@ -447,9 +452,24 @@ class ScpMessage {
            final GenericResp genericResp = GenericResp.fromBuffer(byte);
            return genericResp;
          } else if (messageType == MessageTypeConstant.RealTimeData) {
-            // 没有分包直接解析
-            final TalkData talkData = TalkData.fromBuffer(byte);
-            return talkData;
+            // 回声测试
+            if (spTotal != null &&
+                spTotal > 1 &&
+                messageId != null &&
+                spIndex != null) {
+              // 分包处理
+              return _handleFragmentedPayload(
+                messageId: messageId,
+                spTotal: spTotal,
+                spIndex: spIndex,
+                byte: byte,
+                payloadType: payloadType,
+              );
+            } else {
+              // 没有分包直接解析
+              final TalkData talkData = TalkData.fromBuffer(byte);
+              return talkData;
+            }
          } else {
            String payload = utf8.decode(byte);
            return payload;
@ -470,10 +490,11 @@ class ScpMessage {
          String payload = utf8.decode(byte);
          return payload;
      }
-    } catch (e) {
-      e.printError();
-      _log(text: '❌反序列化udp数据时遇到错误----》$e,message:$messageType byte:${byte.take(20)}');
-      // 尝试打印原始字节数组以供调试
+    } catch (e, stackTrace) {
+      // 打印异常信息
+      _log(text: '❌反序列化udp数据时遇到错误----》$e');
+      // 打印堆栈跟踪信息
+      _log(text: '堆栈跟踪:\n$stackTrace');
      return '';
    }
  }
@ -503,9 +524,10 @@ class ScpMessage {
    required int spTotal,
    required int spIndex,
    required List<int> byte,
+    required int payloadType,
  }) {
-    print(
-        '_handleFragmentedPayload spTotal:$spTotal spindex:$spIndex} byte:${byte.take(20)}');
+    // print(
+    //     '_handleFragmentedPayload spTotal:$spTotal spindex:$spIndex} byte:${byte.take(20)}');
    // 初始化分包列表
    if (!_packetBuffer.containsKey(messageId)) {
      _packetBuffer[messageId] = List.filled(spTotal, []);
@ -523,6 +545,12 @@ class ScpMessage {
      // 清除已重组的分包数据
      _packetBuffer.remove(messageId);

+      if (payloadType == PayloadTypeConstant.talkData) {
+        // 解析完整的 payload
+        final TalkData talkData = TalkData.fromBuffer(completePayload);
+        return talkData;
+      }
+
      return completePayload;

      // 解析完整的 payload
--- a/lib/talk/startChart/handle/impl/udp_talk_accept_handler.dart
+++ b/lib/talk/startChart/handle/impl/udp_talk_accept_handler.dart
@ -27,7 +27,7 @@ class UdpTalkAcceptHandler extends ScpMessageBaseHandle
    final GenericResp genericResp = scpMessage.Payload;
    if (checkGenericRespSuccess(genericResp)) {
      // 延迟2秒后启动监听
-       Future.delayed(Duration(seconds: 3), () {
+       Future.delayed(Duration(seconds: 4), () {
        // 启动通话保持定时器
        _handleStartTalkPing();
        // 启动发送预期数据请求
--- a/lib/talk/startChart/handle/impl/udp_talk_data_handler.dart
+++ b/lib/talk/startChart/handle/impl/udp_talk_data_handler.dart
@ -27,6 +27,7 @@ class UdpTalkDataHandler extends ScpMessageBaseHandle
      final TalkData talkData = scpMessage.Payload;
      // 处理音视频数据
      _handleTalkData(talkData: talkData);
+      print('talkData：$talkData');
      // 收到数据后调用更新，防止定时器超时
      talkDataOverTimeTimerManager.receiveMessage();
    }
--- a/lib/talk/startChart/start_chart_manage.dart
+++ b/lib/talk/startChart/start_chart_manage.dart
@ -50,12 +50,16 @@ class StartChartManage {
  final String _productName = F.navTitle;

  RawDatagramSocket? _udpSocket;
+  final Map<String, Completer<void>> _completers = {}; // 发送消息的请求是否完成
+  final Uuid _uuid = Uuid(); // 用于区分发送的消息的唯一id
+  int _messageMaxTimeout = 5; // 消息最大超时时间（s）
  late String remoteHost = ''; // 远程主机地址(服务器返回)
  late int remotePort = 0; // 远程主机端口(服务器返回)
  final int localPort = 62289; // 本地端口
  String localPublicHost = ''; // 本地公网ip地址

  int heartbeatIntervalTime = 1; // 心跳包间隔时间（s）
+
  Timer? _heartBeatTimer; // 心跳包定时器
  bool _heartBeatTimerRunning = false; // 心跳包定时任务发送状态

@ -308,9 +312,6 @@ class StartChartManage {
      );
      // 发送消息
      await _sendMessage(message: message);
-      // _log(
-      //     text:
-      //         '发送回声测试分包消息=====messageId:$messageId,SpTotal:$totalPackets,SpIndex:${i + 1},packet:${packet.length}');
    }
    // 分包发送完了递增一下id
    MessageCommand.getNextMessageId(toPeerId);
@ -765,8 +766,9 @@ class StartChartManage {
              // _log(text: 'Udp收到结构体数据---》$deserialize');
            }
          }
-        } catch (e) {
+        } catch (e, stackTrace) {
          _log(text: '❌ Udp result data error ----> $e');
+          _log(text: '堆栈跟踪:\n$stackTrace');
        }
      }
    });