rtph265pay元素add_probe()回调未触发，请求排查原因

我尝试给UDP输出的RTP数据包添加RTP头扩展，因此给rtph265pay元素的Pad添加了add_probe()方法，但完全没有触发探针回调。

我的简化GStreamer管道如下：

"rtspsrc location=<myrtspurl> name=rtspsrc latency=100 
      rtspsrc. ! rtph265depay ! tee name=t 
         t. ! queue ! h265parse ! rtph265pay name=rtph265pay config-interval=-1 
 ! udpsink name=udpsink host=127.0.0.1 port=56001";

已验证接收管道能正常播放，说明数据传输正常，但不清楚为何探针无回调。

关键观察：

• 用rtph264pay处理H264 RTSP流时，相同代码能正常触发探针回调，这是H265特有的问题
• 给rtph265pay之后的udpsink的sink Pad添加探针，同样没有回调

我用Rust实现的探针添加代码如下：

rtspsrc.connect_pad_added(move |_, src_pad| {
        match src_pad.current_caps() {
            Some(caps) => {
                let new_pad_struct = caps.structure(0).expect("Failed to get first structure of caps for audio");
                for i in 0..new_pad_struct.n_fields() {
                    match new_pad_struct.nth_field_name(i).unwrap().as_str() {
                        "media" => {
                            let media_type = new_pad_struct.value("media").unwrap();
                            let field_value = media_type.get::<&str>().unwrap();
                            println!("field_value={}", field_value);
                            if field_value == "video" {
                                bin_ref.debug_to_dot_file(gst::DebugGraphDetails::all(), "PLAYING");
                                rtppay_src_pad.add_probe(gst::PadProbeType::BUFFER, |pad, probe_info| {
                                    println!("adding probe for rtppay");
                                    if let Some(probe_data) = probe_info.data.as_mut() {
                                        if let gst::PadProbeData::Buffer(ref mut buffer) = probe_data {
                                            let size = buffer.size();
                                            match buffer.pts() {
                                                Some(pts) => {
                                                    println!("ptstime={}", pts.seconds())
                                                },
                                                None => {
                                                    println!("No PTS, cannot get bandwidth")
                                                }
                                            }

                                            let b = buffer.get_mut().unwrap();
                                            let mut rtp_buffer = RTPBuffer::from_buffer_writable(b).unwrap();

                                            let pts = rtp_buffer.buffer().pts().unwrap();
                                            // Convert the PTS to bytes
                                            let pts_bytes = pts.to_be_bytes();
                                            let extension_data = &pts_bytes[..];

                                            let appbits = 5; // Custom application-specific bits
                                            let id = 1; // Custom extension ID
                                            let result = rtp_buffer.add_extension_twobytes_header(appbits, id, extension_data);

                                            if let Err(e) = result {
                                                eprintln!("Failed to add RTP header extension: {:?}", e);
                                            }
                                        }
                                    }
                                    gst::PadProbeReturn::Ok
                                });

                                udpsink_sink_pad.add_probe(gst::PadProbeType::BUFFER, |pad, probe_info| {
                                    println!("adding pad probe to udpsink");
                                    if let Some(probe_data) = probe_info.data.as_ref() {
                                        if let gst::PadProbeData::Buffer(buffer) = probe_data {
                                            let rtp_buffer = RTPBuffer::from_buffer_readable(buffer).unwrap();
                                            // Check for RTP extension header
                                            if let Some((appbits, extension_data)) = rtp_buffer.extension_twobytes_header(1, 0) { //extension_twobytes_header(1, 0) {
                                                println!("RTP Extension present:");
                                                println!("App bits: {}", appbits);
                                                println!("Extension data: {:?}", extension_data);

                                                // Convert the extension data back to PTS
                                                if extension_data.len() != 0 {
                                                    let mut pts_bytes = [0u8; 8];
                                                    pts_bytes[..4].copy_from_slice(&extension_data[..4]);  // Copy the first 4 bytes
                                                    let pts = u64::from_be_bytes(pts_bytes);
                                                    //let pts = u64::from_be_bytes(extension_data.try_into().unwrap());
                                                    println!("Extracted PTS from RTP extension: {}", pts);
                                                }
                                            } else {
                                                println!("No RTP Extension found");
                                            }
                                            match rtp_buffer.buffer().pts() {
                                                Some(pts) => {
                                                    println!("udpsink buffer.pts={}", pts.seconds());
                                                },
                                                None => {
                                                    println!("No PTS, cannot get bandwidth");
                                                }
                                            }
                                        }
                                    }
                                    gst::PadProbeReturn::Ok
                                }).unwrap();

                            }
                        }
                        _ => {}
                    }
                }
            }
            _ => {}
        }
    });

问题原因及修复方案 #

1. 核心问题：Pad引用的生命周期或匹配错误 #

你代码中rtppay_src_pad和udpsink_sink_pad的获取逻辑缺失，大概率是提前获取了这两个Pad的引用，但此时H265分支的Pad可能还未完全激活，或者获取的Pad并非实际数据流经过的Pad。

在rtspsrc.connect_pad_added回调中，你基于rtspsrc的src pad判断媒体类型，但rtspsrc的视频Pad连接的是rtph265depay，而非rtph265pay，直接引用提前获取的rtppay_src_pad会存在以下问题：

• 当管道动态创建Pad时（比如rtph265pay的src Pad可能延迟创建），提前获取的Pad引用无效
• 引用的Pad未处于数据流路径上

2. 正确的Pad探针添加方式 #

应该在rtph265pay元素的pad-added信号中添加探针，确保捕获到实际激活的src Pad：

// 监听rtph265pay的pad-added信号
rtph265pay.connect_pad_added(move |_, src_pad| {
    if src_pad.name().starts_with("src") {
        src_pad.add_probe(gst::PadProbeType::BUFFER, |pad, probe_info| {
            // 你的RTP扩展添加逻辑
            println!("触发rtph265pay探针回调");
            // ... 原有逻辑 ...
            gst::PadProbeReturn::Ok
        }).unwrap();
    }
});

// 同理监听udpsink的sink pad添加
udpsink.connect_pad_added(move |_, sink_pad| {
    if sink_pad.name().starts_with("sink") {
        sink_pad.add_probe(gst::PadProbeType::BUFFER, |pad, probe_info| {
            // 你的探针逻辑
            println!("触发udpsink探针回调");
            // ... 原有逻辑 ...
            gst::PadProbeReturn::Ok
        }).unwrap();
    }
});

3. 额外排查点 #

• 检查rtph265pay的config-interval=-1是否导致Pad延迟激活：可以尝试将config-interval设为1，强制发送SPS/PPS，触发Pad快速激活
• 确保你的rtppay_src_pad是rtph265pay的src Pad，而非sink Pad：H264和H265的pay元素Pad命名逻辑一致，但如果误绑sink Pad会导致无回调
• 启用GStreamer debug日志：运行时设置GST_DEBUG=*:3，查看Pad的激活状态和数据流路径，确认探针是否正确绑定到数据流经过的Pad

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