GStreamer无法读取RTSP流帧，gst-launch可播放但自研程序报错

我用VLC生成RTSP流，推流命令：
vlc -I dummy -vvv 480.mov --loop --sout '#rtp{sdp=rtsp://localhost:8080/test}'
用gst-launch-1.0 rtspsrc location=rtsp://localhost:8080/test ! decodebin ! autovideosink能正常播放该流，但自行编写的GStreamer C++程序运行报错。

编译命令：
g++ -o rtsp_reader gsttest.cpp $(pkg-config --cflags --libs gstreamer-1.0 gstreamer-video-1.0)

程序运行错误输出：

./rtsp_reader
Playing: rtsp://localhost:8080/test
Error: Internal data stream error.
Returned, stopping playback

程序代码：

#include <gst/gst.h>
#include <stdio.h>

// Define your RTSP URI here
#define RTSP_URI "rtsp://localhost:8080/test"

//vlc -I dummy -vvv 480.mov --loop --sout '#rtp{sdp=rtsp://localhost:8080/test}'

//gst-launch-1.0 rtspsrc location=rtsp://ip/url ! rtph264depay ! h264parse ! mp4mux ! filesink location=file.mp4
//gst-launch-1.0 -e rtspsrc location=rtsp://admin:pass@192.168.85.7/rtsph2641080p protocols=tcp ! rtph264depay ! h264parse ! mp4mux ! filesink location=~/camera.mp4
//gst-launch-1.0 rtspsrc location=rtsp://localhost:8080/test ! decodebin !  autovideosink
//g++ -o rtsp_reader gsttest.cpp $(pkg-config --cflags --libs gstreamer-1.0 gstreamer-video-1.0)

static gboolean bus_call(GstBus *bus, GstMessage *msg, gpointer data) {

    GMainLoop *loop = (GMainLoop *) data;


    switch (GST_MESSAGE_TYPE(msg)) {
        case GST_MESSAGE_EOS:
            g_print("End of stream\n");
            g_main_loop_quit(loop);
            break;

        case GST_MESSAGE_ERROR: {
            gchar *debug;
            GError *error;

            gst_message_parse_error(msg, &error, &debug);
            g_free(debug);

            g_printerr("Error: %s\n", error->message);
            g_error_free(error);

            g_main_loop_quit(loop);
            break;
        }

        default:
            break;
    }

    return TRUE;
}

static GstFlowReturn new_frame(GstElement *sink, gpointer data) {
    GstSample *sample = NULL;
    GstBuffer *buffer = NULL;
    printf("frame \n");
    g_signal_emit_by_name(sink, "pull-sample", &sample);
    
    if (sample) {
        buffer = gst_sample_get_buffer(sample);
        if (buffer) {
            GstCaps *caps = gst_sample_get_caps(sample);
            gint width, height;
            GstStructure * strct = gst_caps_get_structure(caps, 0);
            gst_structure_get_int(strct, "width", &width);
            gst_structure_get_int(strct, "height", &height);
            g_print("Received frame - Width: %d, Height: %d\n", width, height);

        }
        gst_sample_unref(sample);
    }
    return GST_FLOW_OK;
}

int main(int argc, char *argv[]) {
    GstElement *pipeline, *source, *decoder, *sink;
    GstBus *bus;
    GMainLoop *loop;

    /* Initialize GStreamer */
    gst_init(&argc, &argv);
    loop = g_main_loop_new(NULL, FALSE);

    /* Create elements */
    pipeline = gst_pipeline_new("rtsp-pipeline");
    source = gst_element_factory_make("rtspsrc", "source");
    decoder = gst_element_factory_make("decodebin", "decoder");
    sink = gst_element_factory_make("appsink", "sink");

    if (!pipeline || !source || !decoder || !sink) {
        g_printerr("Not all elements could be created.\n");
        return -1;
    }

    /* Set the URI to play */
    g_object_set(source, "location", RTSP_URI, NULL);

    /* Build the pipeline */
    gst_bin_add_many(GST_BIN(pipeline), source, decoder, sink, NULL);
    gst_element_link(source, decoder);
    gst_element_link(decoder, sink);


    /* Set the bus */
    bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
    gst_bus_add_watch(bus, bus_call, loop);
    gst_object_unref(bus);

    /* Set the new sample callback on the sink */
    g_signal_connect(sink, "new-sample", G_CALLBACK(new_frame), NULL);

    /* Start playing */
    g_print("Playing: %s\n", RTSP_URI);
    gst_element_set_state(pipeline, GST_STATE_PLAYING);

    /* Iterate */
    g_main_loop_run(loop);

    /* Out of the main loop, clean up nicely */
    g_print("Returned, stopping playback\n");
    gst_element_set_state(pipeline, GST_STATE_NULL);

    gst_object_unref(GST_OBJECT(pipeline));
    g_main_loop_unref(loop);

    return 0;
}

核心问题 #

1. rtspsrc无法直接和decodebin静态链接
   rtspsrc是多输出元素（可能包含音频、视频等多个流），它的输出pad是动态创建的，不能用gst_element_link直接静态链接到decodebin。gst-launch的!语法会自动处理动态pad链接，但C++程序必须手动监听pad创建事件完成链接。

2. decodebin到appsink的链接同样需要动态处理
   decodebin解码后会根据流类型生成不同的输出pad，也需要监听它的pad-added信号来链接到appsink。

3. appsink缺少必要配置
   默认appsink未设置合适的caps，导致数据流不匹配，需要明确指定视频格式的caps（如video/x-raw）。

修复后的代码 #

#include <gst/gst.h>
#include <stdio.h>

#define RTSP_URI "rtsp://localhost:8080/test"

static gboolean bus_call(GstBus *bus, GstMessage *msg, gpointer data) {
    GMainLoop *loop = (GMainLoop *)data;

    switch (GST_MESSAGE_TYPE(msg)) {
        case GST_MESSAGE_EOS:
            g_print("End of stream\n");
            g_main_loop_quit(loop);
            break;
        case GST_MESSAGE_ERROR: {
            gchar *debug;
            GError *error;
            gst_message_parse_error(msg, &error, &debug);
            g_free(debug);
            g_printerr("Error: %s\n", error->message);
            g_error_free(error);
            g_main_loop_quit(loop);
            break;
        }
        default:
            break;
    }
    return TRUE;
}

static GstFlowReturn new_frame(GstElement *sink, gpointer data) {
    GstSample *sample = NULL;
    g_signal_emit_by_name(sink, "pull-sample", &sample);
    
    if (sample) {
        GstBuffer *buffer = gst_sample_get_buffer(sample);
        if (buffer) {
            GstCaps *caps = gst_sample_get_caps(sample);
            gint width, height;
            GstStructure *strct = gst_caps_get_structure(caps, 0);
            if (gst_structure_get_int(strct, "width", &width) && gst_structure_get_int(strct, "height", &height)) {
                g_print("Received frame - Width: %d, Height: %d\n", width, height);
            }
        }
        gst_sample_unref(sample);
    }
    return GST_FLOW_OK;
}

// 处理decodebin的pad添加事件，链接到appsink
static void decodebin_pad_added(GstElement *decodebin, GstPad *new_pad, gpointer data) {
    GstElement *sink = (GstElement *)data;
    GstPad *sink_pad = gst_element_get_static_pad(sink, "sink");
    
    // 检查pad是否已经链接
    if (gst_pad_is_linked(sink_pad)) {
        g_print("Sink pad already linked, skipping\n");
        gst_object_unref(sink_pad);
        return;
    }

    // 获取新pad的caps
    GstCaps *caps = gst_pad_get_current_caps(new_pad);
    GstStructure *strct = gst_caps_get_structure(caps, 0);
    const gchar *media_type = gst_structure_get_name(strct);

    // 只处理视频流
    if (!g_str_has_prefix(media_type, "video/")) {
        g_print("Ignoring non-video pad: %s\n", media_type);
        gst_caps_unref(caps);
        gst_object_unref(sink_pad);
        return;
    }

    // 链接pad
    GstPadLinkReturn ret = gst_pad_link(new_pad, sink_pad);
    if (GST_PAD_LINK_FAILED(ret)) {
        g_printerr("Failed to link decodebin pad to appsink\n");
    } else {
        g_print("Successfully linked decodebin pad to appsink\n");
    }

    gst_caps_unref(caps);
    gst_object_unref(sink_pad);
}

// 处理rtspsrc的pad添加事件，链接到decodebin
static void rtspsrc_pad_added(GstElement *rtspsrc, GstPad *new_pad, gpointer data) {
    GstElement *decodebin = (GstElement *)data;
    GstPad *decodebin_pad = gst_element_get_static_pad(decodebin, "sink");
    
    if (gst_pad_is_linked(decodebin_pad)) {
        g_print("Decodebin sink pad already linked, skipping\n");
        gst_object_unref(decodebin_pad);
        return;
    }

    GstPadLinkReturn ret = gst_pad_link(new_pad, decodebin_pad);
    if (GST_PAD_LINK_FAILED(ret)) {
        g_printerr("Failed to link rtspsrc pad to decodebin\n");
    } else {
        g_print("Successfully linked rtspsrc pad to decodebin\n");
    }

    gst_object_unref(decodebin_pad);
}

int main(int argc, char *argv[]) {
    GstElement *pipeline, *source, *decoder, *sink;
    GstBus *bus;
    GMainLoop *loop;

    gst_init(&argc, &argv);
    loop = g_main_loop_new(NULL, FALSE);

    pipeline = gst_pipeline_new("rtsp-pipeline");
    source = gst_element_factory_make("rtspsrc", "source");
    decoder = gst_element_factory_make("decodebin", "decoder");
    sink = gst_element_factory_make("appsink", "sink");

    if (!pipeline || !source || !decoder || !sink) {
        g_printerr("Not all elements could be created.\n");
        return -1;
    }

    // 设置rtspsrc的URI
    g_object_set(source, "location", RTSP_URI, NULL);
    // 可选：设置rtspsrc使用TCP协议，避免UDP丢包问题
    // g_object_set(source, "protocols", GST_RTSP_LOWER_TRANS_TCP, NULL);

    // 配置appsink：设置caps为视频原始格式，启用异步模式
    GstCaps *sink_caps = gst_caps_new_simple("video/x-raw",
        "format", G_TYPE_STRING, "RGB",
        NULL);
    g_object_set(sink, "caps", sink_caps, "emit-signals", TRUE, "sync", FALSE, NULL);
    gst_caps_unref(sink_caps);

    // 添加元素到管道
    gst_bin_add_many(GST_BIN(pipeline), source, decoder, sink, NULL);

    // 监听pad添加信号，动态链接元素
    g_signal_connect(source, "pad-added", G_CALLBACK(rtspsrc_pad_added), decoder);
    g_signal_connect(decoder, "pad-added", G_CALLBACK(decodebin_pad_added), sink);

    // 设置bus监听
    bus = gst_pipeline_get_bus(GST_PIPELINE(pipeline));
    gst_bus_add_watch(bus, bus_call, loop);
    gst_object_unref(bus);

    // 设置appsink的新样本回调
    g_signal_connect(sink, "new-sample", G_CALLBACK(new_frame), NULL);

    // 启动管道
    g_print("Playing: %s\n", RTSP_URI);
    gst_element_set_state(pipeline, GST_STATE_PLAYING);

    // 运行主循环
    g_main_loop_run(loop);

    // 清理资源
    g_print("Returned, stopping playback\n");
    gst_element_set_state(pipeline, GST_STATE_NULL);
    gst_object_unref(GST_OBJECT(pipeline));
    g_main_loop_unref(loop);

    return 0;
}

关键修复点说明 #

• 动态Pad链接：通过监听rtspsrc和decodebin的pad-added信号，在pad创建时完成动态链接，这是处理多输出元素的标准方式。
• Appsink配置：设置video/x-raw的caps确保只接收解码后的视频原始数据，开启emit-signals让new-sample信号正常触发，关闭sync避免阻塞。
• 可选优化：给rtspsrc设置protocols为GST_RTSP_LOWER_TRANS_TCP，使用TCP协议传输RTSP流，避免UDP可能的丢包问题，提升稳定性。

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