sky/app-starlock
3
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

develop_sky #1

Merged
liyi merged 62 commits from develop_sky into develop_sky_liyi 2025-09-22 17:48:20 +08:00
Conversation 0 Commits 62 Files Changed 2 +123 -206
9 changed files with 123 additions and 206 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files
Showing only changes of commit 954cfc6276 - Show all commits

9
lib/talk/starChart/handle/impl/udp_talk_data_handler.dart
View File

@ -16,6 +16,11 @@ import 'package:star_lock/talk/starChart/proto/talk_data_h264_frame.pb.dart';
// implements ScpMessageHandler {
class UdpTalkDataHandler extends ScpMessageBaseHandle
implements ScpMessageHandler {
// 单例实现
static final UdpTalkDataHandler instance = UdpTalkDataHandler();
UdpTalkDataHandler(); // 保持默认构造函数
@override
void handleReq(ScpMessage scpMessage) {}
@ -32,7 +37,7 @@ class UdpTalkDataHandler extends ScpMessageBaseHandle
if (scpMessage.Payload != null) {
final TalkData talkData = scpMessage.Payload;
// 处理音视频数据
_handleTalkData(
talkData: talkData,
scpMessage: scpMessage,
@ -122,7 +127,9 @@ class UdpTalkDataHandler extends ScpMessageBaseHandle
void _handleVideoH264(TalkData talkData, ScpMessage scpMessage) {
final TalkDataH264Frame talkDataH264Frame = TalkDataH264Frame();
talkDataH264Frame.mergeFromBuffer(talkData.content);
// AppLog.log('处理H264帧: frameType=${talkDataH264Frame.frameType}, frameSeq=${talkDataH264Frame.frameSeq},MessageId:${scpMessage.MessageId}');
frameHandler.handleFrame(talkDataH264Frame, talkData, scpMessage);
}
/// 处理图片数据

3
lib/talk/starChart/handle/other/h264_frame_handler.dart
View File

@ -15,6 +15,9 @@ class H264FrameHandler {
void handleFrame(
TalkDataH264Frame frame, TalkData talkData, ScpMessage scpMessage) {
// AppLog.log(
// '送入stream的帧数据: frameSeq=${frame.frameSeq},frameType=${frame
// .frameType},MessageId:${scpMessage.MessageId}');
onCompleteFrame(TalkDataModel(
talkData: talkData,
talkDataH264Frame: frame,

2
lib/talk/starChart/handle/other/talk_data_repository.dart
View File

@ -11,7 +11,7 @@ class TalkDataRepository {
onCancel: () {
_isListening = false;
},
sync: false, // 改为同步模式以提高实时性
sync: true, // 改为同步模式以提高实时性
);
}

34
lib/talk/starChart/handle/scp_message_base_handle.dart
View File

@ -181,40 +181,6 @@ class ScpMessageBaseHandle {
}
return null;
// if (!_packetBuffer.containsKey(key)) {
// _packetBuffer[key] = List.filled(spTotal, []);
// _startTimer(key);
// }
//
// // 检查分包索引是否在合法范围内
// if (spIndex < 1 || spIndex > spTotal) {
// // print(
// // 'Invalid spTotal: $spTotal spIndex: $spIndex for messageId: $messageId');
// return null;
// }
//
// // 存储当前分包
// _packetBuffer[key]![spIndex - 1] = byte;
//
// // 检查是否接收到所有分包
// if (_packetBuffer[key]!.every((packet) => packet.isNotEmpty)) {
// // 重组所有分包
// Uint8List completePayload = Uint8List.fromList(
// _packetBuffer[key]!.expand((packet) => packet).toList());
// // 清除已重组和超时的分包数据
// _clearPacketData(key);
//
// // 使用重组的包构造成TalkData
// if (payloadType == PayloadTypeConstant.talkData) {
// final talkData = TalkData();
// talkData.mergeFromBuffer(completePayload);
// return talkData;
// }
// } else {
// // 如果分包尚未接收完全，返回 null 或其他指示符
// return null;
// }
}
// 启动定时器

2
lib/talk/starChart/handle/scp_message_handler_factory.dart
View File

@ -52,7 +52,7 @@ class ScpMessageHandlerFactory {
case PayloadTypeConstant.talkExpect:
return UdpTalkExpectHandler();
case PayloadTypeConstant.talkData:
return UdpTalkDataHandler();
return UdpTalkDataHandler.instance;
case PayloadTypeConstant.talkHangup:
return UdpTalkHangUpHandler();
case PayloadTypeConstant.RbcuInfo:

115
lib/talk/starChart/star_chart_manage.dart
View File

@ -141,11 +141,6 @@ class StartChartManage {
return _avFrameLost / _avFrameTotal;
}
// ====== Isolate相关成员变量 ======
Isolate? _udpIsolate;
SendPort? _udpIsolateSendPort;
ReceivePort? _mainReceivePort;
// 星图服务初始化
Future<void> init() async {
if (F.isXHJ) {
@ -167,8 +162,6 @@ class StartChartManage {
await _onlineRelayService();
// 上报
await reportInformation();
// 初始化Isolate
await _initUdpIsolate();
}
/// 客户端注册
@ -248,21 +241,20 @@ class StartChartManage {
var addressIListenFrom = InternetAddress.anyIPv4;
RawDatagramSocket.bind(addressIListenFrom, localPort)
.then((RawDatagramSocket socket) {
// 设置接收缓冲区大小 (SO_RCVBUF = 8)
socket.setRawOption(
RawSocketOption.fromInt(
RawSocketOption.levelSocket,
8, // SO_RCVBUF for Android/iOS
8, // SO_RCVBUF for Android/iOS
2 * 1024 * 1024, // 2MB receive buffer
),
);
// 设置发送缓冲区大小 (SO_SNDBUF = 7)
socket.setRawOption(
RawSocketOption.fromInt(
RawSocketOption.levelSocket,
7, // SO_SNDBUF for Android/iOS
7, // SO_SNDBUF for Android/iOS
2 * 1024 * 1024, // 2MB send buffer
),
);
@ -1057,29 +1049,22 @@ class StartChartManage {
// 接收返回的数据
void _onReceiveData(RawDatagramSocket socket, BuildContext context) {
// ====== Isolate初始化握手 ======
if (_udpIsolateSendPort == null && _mainReceivePort != null) {
_mainReceivePort!.listen((dynamic message) {
if (message is SendPort) {
_udpIsolateSendPort = message;
}
});
}
socket.listen((RawSocketEvent event) {
if (event == RawSocketEvent.read) {
Datagram? dg;
while ((dg = socket.receive()) != null) {
try {
if (dg?.data != null) {
// 直接将bytes发送到Isolate
if (_udpIsolateSendPort != null) {
_udpIsolateSendPort!.send(dg!.data);
} else {
// Fallback: 主线程处理（极端情况）
final deserialize = ScpMessage.deserialize(dg!.data);
if (deserialize != null) {
_handleUdpResultData(deserialize);
}
// Fallback: 主线程处理（极端情况）
// 更好的做法：批量处理
final deserialize = ScpMessage.deserialize(dg!.data);
// if (deserialize.PayloadType == PayloadTypeConstant.talkData) {
// _log(
// text: 'mesaageId:${deserialize.MessageId},'
// 'SpTotal:${deserialize.SpTotal},SpIndex:${deserialize.SpIndex}');
// }
if (deserialize != null) {
_handleUdpResultData(deserialize);
}
}
} catch (e, stackTrace) {
@ -1095,14 +1080,6 @@ class StartChartManage {
final int payloadType = scpMessage.PayloadType ?? 0;
final int messageType = scpMessage.MessageType ?? 0;
try {
// 记录分包数据用于统计丢包率
// if (scpMessage.SpIndex != null &&
// scpMessage.SpTotal != null &&
// scpMessage.MessageId != null) {
// PacketLossStatistics().recordPacket(
// scpMessage.MessageId!, scpMessage.SpIndex!, scpMessage.SpTotal!);
// }
final ScpMessageHandler handler =
ScpMessageHandlerFactory.createHandler(payloadType);
if (messageType == MessageTypeConstant.Req) {
@ -1303,8 +1280,6 @@ class StartChartManage {
await Storage.removerStarChartRegisterNodeInfo();
// 关闭udp服务
closeUdpSocket();
// 关闭Isolate
_disposeUdpIsolate();
}
/// 重置数据
@ -1313,68 +1288,4 @@ class StartChartManage {
isOnlineStarChartServer = false;
talkStatus.setUninitialized();
}
// 初始化Isolate
Future<void> _initUdpIsolate() async {
_mainReceivePort = ReceivePort();
_mainReceivePort!.listen((dynamic message) {
// 目前不需要主线程回调，如需日志可在此处理
// if (message is String) {
// _log(text: '[Isolate] $message');
// }
});
_udpIsolate = await Isolate.spawn<_UdpIsolateInitParams>(
_udpIsolateEntry,
_UdpIsolateInitParams(
_mainReceivePort!.sendPort,
_handleUdpResultDataStatic,
),
debugName: 'UdpDataIsolate',
);
}
// 销毁Isolate
void _disposeUdpIsolate() {
_udpIsolateSendPort = null;
_mainReceivePort?.close();
_mainReceivePort = null;
_udpIsolate?.kill(priority: Isolate.immediate);
_udpIsolate = null;
}
// 静态代理，便于Isolate调用成员方法
static void _handleUdpResultDataStatic(Map<String, dynamic> args) {
final instance = StartChartManage();
final ScpMessage scpMessage = args['scpMessage'] as ScpMessage;
instance._handleUdpResultData(scpMessage);
}
// Isolate入口函数
static void _udpIsolateEntry(_UdpIsolateInitParams params) {
final SendPort mainSendPort = params.mainSendPort;
final Function handleUdpResultData = params.handleUdpResultData;
final ReceivePort isolateReceivePort = ReceivePort();
// 首次将SendPort发回主线程
mainSendPort.send(isolateReceivePort.sendPort);
isolateReceivePort.listen((dynamic message) {
try {
if (message is List<int>) {
// 修正：List<int>转Uint8List
final scpMessage = ScpMessage.deserialize(Uint8List.fromList(message));
if (scpMessage != null) {
// 通过静态代理调用主线程的_handleUdpResultData
handleUdpResultData({'scpMessage': scpMessage});
}
}
} catch (e, stack) {
// 可选：mainSendPort.send('[Isolate Error] $e\n$stack');
}
});
}
}
class _UdpIsolateInitParams {
final SendPort mainSendPort;
final Function handleUdpResultData;
_UdpIsolateInitParams(this.mainSendPort, this.handleUdpResultData);
}

142
lib/talk/starChart/views/native/talk_view_native_decode_logic.dart
View File

@ -38,6 +38,7 @@ import 'package:star_lock/tools/G711Tool.dart';
import 'package:star_lock/tools/bugly/bugly_tool.dart';
import 'package:star_lock/tools/commonDataManage.dart';
import 'package:star_lock/tools/storage.dart';
import 'package:video_decode_plugin/nalu_utils.dart';
import 'package:video_decode_plugin/video_decode_plugin.dart';
import '../../../../tools/baseGetXController.dart';
@ -51,6 +52,15 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
int audioBufferSize = 2; // 音频默认缓冲2帧
// 回绕阈值，动态调整，frameSeq较小时阈值也小
int _getFrameSeqRolloverThreshold(int lastSeq) {
if (lastSeq > 2000) {
return 1000;
} else {
return (lastSeq / 2).round();
}
}
// 定义音频帧缓冲和发送函数
final List<int> _bufferedAudioFrames = <int>[];
@ -162,28 +172,35 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
int frameSeqI,
ScpMessage scpMessage,
) {
// 检测frameSeq回绕，且为I帧
// 动态回绕阈值判断，frameSeq较小时阈值也小
if (!_pendingStreamReset &&
_lastFrameSeq != null &&
frameType == TalkDataH264Frame_FrameTypeE.I &&
frameSeq < _lastFrameSeq!) {
// 检测到新流I帧，进入loading并重置所有本地状态
AppLog.log(
'检测到新流I帧，frameSeq回绕，进入loading并重置: frameSeq=$frameSeq, lastFrameSeq=$_lastFrameSeq');
Future.microtask(() => state.isLoading.value = true);
_pendingStreamReset = true;
// 先暂停帧处理定时器，防止竞态
_stopFrameProcessTimer();
// 先释放并重新初始化解码器
_resetDecoderForNewStream(_pendingResetWidth, _pendingResetHeight);
// 重置所有本地状态
_lastFrameSeq = null;
_decodedIFrames.clear();
state.h264FrameBuffer.clear();
// 再恢复帧处理定时器
_startFrameProcessTimer();
// 不return，直接用该I帧初始化解码器并解码
// 继续往下执行
int dynamicThreshold = _getFrameSeqRolloverThreshold(_lastFrameSeq!);
if ((_lastFrameSeq! - frameSeq) > dynamicThreshold) {
// 检测到新流I帧，frameSeq大幅回绕，进入loading并重置所有本地状态
AppLog.log('检测到新流I帧，frameSeq大幅回绕，进入loading并重置: frameSeq=$frameSeq, lastFrameSeq=$_lastFrameSeq, 阈值=$dynamicThreshold');
Future.microtask(() => state.isLoading.value = true);
_pendingStreamReset = true;
// 先暂停帧处理定时器，防止竞态
_stopFrameProcessTimer();
// 先释放并重新初始化解码器
_resetDecoderForNewStream(_pendingResetWidth, _pendingResetHeight);
// 重置所有本地状态
_lastFrameSeq = null;
_decodedIFrames.clear();
state.h264FrameBuffer.clear();
// 再恢复帧处理定时器
_startFrameProcessTimer();
// 不return，直接用该I帧初始化解码器并解码
// 继续往下执行
} else {
// 小幅度乱序，直接丢弃
AppLog.log('检测到I帧乱序（未超过回绕阈值$dynamicThreshold），丢弃: frameSeq=$frameSeq, lastFrameSeq=$_lastFrameSeq');
return;
}
}
// 如果处于pendingStreamReset，等待新I帧
if (_pendingStreamReset) {
@ -306,6 +323,9 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
lastDecodedIFrameSeq = minIFrameSeq;
// AppLog.log('送入解码器的P帧数据frameSeq:${frameSeq},frameSeqI:${frameSeqI},'
// 'frameType:${frameType},messageId:${scpMessage!.MessageId}');
// final spsData = NaluUtils.filterNalusByType(frameData, 7);
// final ppsData = NaluUtils.filterNalusByType(frameData, 8);
// AppLog.log('SPSDATA:${spsData}，ppsData:${ppsData}');
await VideoDecodePlugin.sendFrame(
frameData: frameData,
frameType: 0,
@ -360,8 +380,9 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
state.isProcessingFrame = false;
return;
}
// AppLog.log('送入解码器的P帧数据frameSeq:${frameSeq},frameSeqI:${frameSeqI},'
// AppLog.log('送入解码器的I帧数据frameSeq:${frameSeq},frameSeqI:${frameSeqI},'
// 'frameType:${frameType},messageId:${scpMessage!.MessageId}');
await VideoDecodePlugin.sendFrame(
frameData: frameData,
frameType: 1,
@ -392,6 +413,11 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
// 监听音视频数据流
void _startListenTalkData() {
// 取消旧订阅
if (_streamSubscription != null) {
_streamSubscription!.cancel();
_streamSubscription = null;
}
// 防止重复监听
if (_isListening) {
AppLog.log("已经存在数据流监听，避免重复监听");
@ -403,39 +429,7 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
_streamSubscription = state.talkDataRepository.talkDataStream
.listen((TalkDataModel talkDataModel) async {
final talkData = talkDataModel.talkData;
final talkDataH264Frame = talkDataModel.talkDataH264Frame;
final scpMessage = talkDataModel.scpMessage;
final contentType = talkData!.contentType;
// 判断数据类型，进行分发处理
switch (contentType) {
case TalkData_ContentTypeE.G711:
if (state.audioBuffer.length >= audioBufferSize) {
state.audioBuffer.removeAt(0); // 丢弃最旧的数据
}
state.audioBuffer.add(talkData); // 添加新数据
// 添加音频播放逻辑，与视频类似
_playAudioFrames();
break;
case TalkData_ContentTypeE.H264:
// 处理H264帧
if (state.textureId.value != null || true) {
if (talkDataH264Frame != null) {
_addFrameToBuffer(
talkData.content,
talkDataH264Frame.frameType,
talkData.durationMs,
talkDataH264Frame.frameSeq,
talkDataH264Frame.frameSeqI,
scpMessage!,
);
}
} else {
AppLog.log('无法处理H264帧：textureId为空');
}
break;
}
_processFrame(talkDataModel);
});
}
@ -1372,7 +1366,7 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
try {
// 先停止帧处理定时器
_stopFrameProcessTimer();
// 释放旧解码器
if (state.textureId.value != null) {
await VideoDecodePlugin.releaseDecoder();
@ -1381,7 +1375,7 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
// 等待一小段时间确保资源释放完成
await Future.delayed(Duration(milliseconds: 100));
// 创建新的解码器配置
final config = VideoDecoderConfig(
width: width,
@ -1394,7 +1388,7 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
if (textureId != null) {
state.textureId.value = textureId;
AppLog.log('frameSeq回绕后解码器初始化成功：textureId=$textureId');
// 设置帧渲染监听
VideoDecodePlugin.setOnFrameRenderedListener((textureId) {
AppLog.log('已经开始渲染=======');
@ -1404,7 +1398,7 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
// 重新启动帧处理定时器
_startFrameProcessTimer();
// 重置相关状态
_decodedIFrames.clear();
state.h264FrameBuffer.clear();
@ -1422,4 +1416,40 @@ class TalkViewNativeDecodeLogic extends BaseGetXController {
state.isLoading.value = false;
}
}
void _processFrame(TalkDataModel talkDataModel) {
final talkData = talkDataModel.talkData;
final talkDataH264Frame = talkDataModel.talkDataH264Frame;
final scpMessage = talkDataModel.scpMessage;
final contentType = talkData!.contentType;
// 判断数据类型，进行分发处理
switch (contentType) {
case TalkData_ContentTypeE.G711:
if (state.audioBuffer.length >= audioBufferSize) {
state.audioBuffer.removeAt(0); // 丢弃最旧的数据
}
state.audioBuffer.add(talkData); // 添加新数据
// 添加音频播放逻辑，与视频类似
_playAudioFrames();
break;
case TalkData_ContentTypeE.H264:
// 处理H264帧
if (state.textureId.value != null) {
if (talkDataH264Frame != null) {
_addFrameToBuffer(
talkData.content,
talkDataH264Frame.frameType,
talkData.durationMs,
talkDataH264Frame.frameSeq,
talkDataH264Frame.frameSeqI,
scpMessage!,
);
}
} else {
AppLog.log('无法处理H264帧：textureId为空');
}
break;
}
}
}

20
lib/talk/starChart/views/native/talk_view_native_decode_page.dart
View File

@ -114,16 +114,16 @@ class _TalkViewNativeDecodePageState extends State<TalkViewNativeDecodePage>
key: state.globalKey,
child: SizedBox.expand(
child: RotatedBox(
// 解码器不支持硬件旋转，使用RotatedBox
quarterTurns: startChartManage.rotateAngle ~/ 90,
child: Platform.isIOS
? Transform.scale(
scale: 1.008, // 轻微放大，消除iOS白边
child: Texture(
textureId: state.textureId.value!,
filterQuality: FilterQuality.medium,
),
)
// 解码器不支持硬件旋转，使用RotatedBox
quarterTurns: startChartManage.rotateAngle ~/ 90,
child: Platform.isIOS
? Transform.scale(
scale: 1.008, // 轻微放大，消除iOS白边
child: Texture(
textureId: state.textureId.value!,
filterQuality: FilterQuality.medium,
),
)
: Texture(
textureId: state.textureId.value!,
filterQuality: FilterQuality.medium,

2
lib/talk/starChart/views/native/talk_view_native_decode_state.dart
View File

@ -110,7 +110,7 @@ class TalkViewNativeDecodeState {
// H264帧缓冲区相关
final List<Map<String, dynamic>> h264FrameBuffer = <Map<String, dynamic>>[]; // H264帧缓冲区，存储帧数据和类型
final int maxFrameBufferSize = 150; // 最大缓冲区大小
final int maxFrameBufferSize = 50; // 最大缓冲区大小
final int targetFps = 60; // 目标解码帧率,只是为了快速填充native的缓冲区
Timer? frameProcessTimer; // 帧处理定时器
bool isProcessingFrame = false; // 是否正在处理帧