GStreamer无法处理appsrc传入的u8向量中TS分段的问题排查

核心错误是appsrc的Caps配置完全错误：你给appsrc设置了video/x-h264格式，但后续接的tsdemux需要的是MPEG-TS流（格式为video/mpegts）——和你用gst-launch测试的逻辑一致，原命令里filesrc输出的是TS文件的原始字节流，对应video/mpegts格式，而非裸H264。格式不匹配直接导致tsdemux无法解析输入，触发"Internal data stream error"。

其次，你的need_data回调逻辑存在问题：每次触发都会重新读取整个文件并推送，导致重复发送完整TS流，进一步加剧解析错误。

修正步骤 #

1. 修正appsrc的Caps与格式配置
   将appsrc的Caps改为video/mpegts，同时把格式设为Bytes更适合处理文件字节流，避免时间戳相关问题：

   let ts_caps = &gst::Caps::builder("video/mpegts")
       .build();
   
   let appsrc = gst_app::AppSrc::builder()
       .caps(ts_caps)
       .format(gst::Format::Bytes)
       .build();
   

2. 修复文件读取与推送逻辑
   不要在need_data里重复读取整个文件，改为一次性读取后推送，最后发送EOS（流结束标记）：

   appsrc.set_callbacks(
       gst_app::AppSrcCallbacks::builder()
           .need_data(move |appsrc, _| {
               static mut HAS_SENT: bool = false;
               unsafe {
                   if HAS_SENT {
                       let _ = appsrc.end_of_stream();
                       return;
                   }
                   HAS_SENT = true;
               }
   
               let mut file = match File::open("./dip.ts") {
                   Ok(f) => f,
                   Err(e) => {
                       eprintln!("读取文件失败: {}", e);
                       let _ = appsrc.end_of_stream();
                       return;
                   }
               };
               let mut buffer = Vec::new();
               if let Err(e) = file.read_to_end(&mut buffer) {
                   eprintln!("读取文件内容失败: {}", e);
                   let _ = appsrc.end_of_stream();
                   return;
               }
   
               let mut gst_buffer = gst::Buffer::from_slice(buffer);
               // 一次性推送时设置不连续标记
               let buffer_ref = gst_buffer.get_mut().unwrap();
               buffer_ref.set_flags(gst::BufferFlags::DISCONT);
   
               match appsrc.push_buffer(gst_buffer) {
                   Ok(_) => println!("推送buffer成功"),
                   Err(e) => eprintln!("推送buffer失败: {}", e),
               }
           })
           .build(),
   );
   

   如果要模拟从RabbitMQ分批接收的场景，可以按TS标准包大小（188字节）分块推送：

   // 分块推送示例，假设buffer是从RabbitMQ获取的u8向量
   const TS_PACKET_SIZE: usize = 188;
   let mut is_first_chunk = true;
   for chunk in buffer.chunks(TS_PACKET_SIZE) {
       let mut gst_buffer = gst::Buffer::from_slice(chunk.to_vec());
       if is_first_chunk {
           let buffer_ref = gst_buffer.get_mut().unwrap();
           buffer_ref.set_flags(gst::BufferFlags::DISCONT);
           is_first_chunk = false;
       }
       if let Err(e) = appsrc.push_buffer(gst_buffer) {
           eprintln!("推送块失败: {}", e);
           break;
       }
   }
   let _ = appsrc.end_of_stream();
   

3. 可选：简化Pad链接逻辑
   原代码手动处理pad链接的逻辑可行，但可以用link_pads_filtered自动匹配视频Pad，更简洁：

   // 替换原有的connect_pad_added逻辑
   tsdemux.link_pads_filtered(Some("video_%u"), &h264parse, Some("sink"))?;
   

修正后，管道逻辑与gst-launch命令完全一致，应该可以正常运行。

内容的提问来源于stack exchange，提问作者Gatothgaj