GStreamer 03 Dynamic pipelines

一、目标

本篇的目标是在应用程序中动态地构建管线，而不是在一开始的程序中构建一个固定的管线。本篇的知识要点有：

如何在连接元素时获得更好的控制。
如何获取感兴趣的事件。
一个元素可用的不用状态。

二、简介

在本篇中，管线在被设置为播放状态前并未被完全构建。如果我们不采取进一步的行动，数据将到达管线的末端，管线将产生一条错误消息并停止。但我们将采取进一步行动。

在此示例中，我们打开一个多路复用（或多路复用）文件，即音频和视频一起存储在容器文件中。负责打开此类容器的元素称为分离器，容器格式的一些示例包括Matroska (MKV)、Quick Time（QT、MOV）、Ogg或高级系统格式（ASF、WMV、WMA）。

GStreamer元素相互通信的端口称为pads(GstPad)。有sink pads，数据通过它进入元素；还有source pads，数据通过它退出元素。很自然地，source元素仅包含source pad，sink元素仅包含sink pad，而filter元素包含两者。

一个多路复用器有一个sink pad和多个src pads。

为了便于理解，下面是一个播放视频的管线，一个demuxer有两个分支，一个是音频流、一个是视频流。该图片与本篇的代码所构建的管线并不是同一个（本篇构建的是一个视频播放器，而GStreamer的对应教程构建的是一个音频播放器）。

demuxer的主要复杂性在于：它们在收到一些数据并有机会查看容器内容之前无法生成任何信息。也就是说，demuxer开始时没有其他元素可以链接到的source pads，因此管线必须在它们处终止。

解决方案是构建从source到demuxer的管线，并将其设置为运行（播放）。当demuxer收到足够的信息来了解容器中流的数量和种类时，它将开始创建source pad。这是我们完成管线构建，并将其附加到新添加的demuxer pad的正确时间。

三、代码

下面是文章01中自动构建管线的命令：

gst-launch-1.0 v4l2src device=/dev/video0 !video/x-raw, format=NV12, width=640, height=480, framerate=30/1 ! autovideosink

基于这条命令，我们现在手动构建管线，代码如下：

static GstElement *pipeline;
static GstElement *source, *demuxer, *queue, *parser, *decoder, *converter, *sink;

static void linkElements(GstElement* element, GstPad* sourcePad, gpointer sinkElement)
{
    GstPad* sinkPad = gst_element_get_static_pad((GstElement*)sinkElement, "sink");
    gst_pad_link(sourcePad, sinkPad);
    gst_object_unref(sinkPad);
}

int MyTest(int argc, char** argv)
{
    GstBus *bus;
    GstMessage *msg;
    GstStateChangeReturn ret;
    gboolean terminate = FALSE;

    /* Initialize GStreamer */
    gst_init(&argc, &argv);

    /* Create the elements */
    source = gst_element_factory_make("filesrc", "file-source");
    demuxer = gst_element_factory_make("qtdemux", "demuxer");
    queue = gst_element_factory_make("queue", "queue");
    parser = gst_element_factory_make("h264parse", "parser");
    decoder = gst_element_factory_make("mppvideodec", "decoder");
    converter = gst_element_factory_make("videoconvert", "converter");
    sink = gst_element_factory_make("autovideosink", "sink");

    /* Create the empty pipeline */
    pipeline = gst_pipeline_new("test-pipeline");

    /* Check creation */
    if (!pipeline || !source || !demuxer || !queue || !parser || !decoder || !converter || !sink) {
        g_printerr("Not all elements could be created.\n");
        return -1;
    }

    /* Set the location of the input file */
    g_object_set(G_OBJECT(source), "location", "13850_h264.mp4", NULL);

    /* Build the pipeline */
    gst_bin_add_many(GST_BIN(pipeline), source, demuxer, queue, parser, decoder, converter, sink, NULL);

    /* Link */
    if (!gst_element_link(source, demuxer)) {
        g_printerr("Source and demuxer could not be linked.\n");
        gst_object_unref(pipeline);
        return -1;
    }
    if (!gst_element_link(queue, parser)) {
        g_printerr("Queue and parser could not be linked.\n");
        gst_object_unref(pipeline);
        return -1;
    }
    if (!gst_element_link(parser, decoder)) {
        g_printerr("Parser and decoder could not be linked.\n");
        gst_object_unref(pipeline);
        return -1;
    }
    if (!gst_element_link(decoder, converter)) {
        g_printerr("Decoder and converter could not be linked.\n");
        gst_object_unref(pipeline);
        return -1;
    }

    /* Manually connect the demuxer to the queue */
    g_signal_connect(demuxer, "pad-added", G_CALLBACK(linkElements), queue);

    /* Link */
    if (!gst_element_link(converter, sink)) {
        g_printerr("Converter and sink could not be linked.\n");
        gst_object_unref(pipeline);
        return -1;
    }

    /* Set the pipeline to "playing" state */
    ret = gst_element_set_state(pipeline, GST_STATE_PLAYING);
    if (ret == GST_STATE_CHANGE_FAILURE) {
        g_printerr("Unable to set the pipeline to the playing state.\n");
        gst_object_unref(pipeline);
        return -1;
    }

    /* Bus */
    bus = gst_element_get_bus(pipeline);
    msg = gst_bus_timed_pop_filtered(
        bus, GST_CLOCK_TIME_NONE,
        GST_MESSAGE_ERROR | GST_MESSAGE_EOS
    );

    /* Parse message */
    if (msg != NULL)
    {
        GError *err;
        gchar *debug_info;

        switch (GST_MESSAGE_TYPE(msg))
        {
            case GST_MESSAGE_ERROR:
                gst_message_parse_error(msg, &err, &debug_info);

                g_printerr("Error received from element %s: %s\n",
                    GST_OBJECT_NAME (msg->src), err->message
                );
                g_printerr("Debugging information: %s\n",
                    debug_info ? debug_info : "none"
                );

                g_clear_error(&err);
                g_free(debug_info);
                break;
            case GST_MESSAGE_EOS:
                g_print("End-Of-Stream reached.\n");
                break;
            default:
                g_printerr("Unexpected message received.\n");
                break;
        }

        gst_object_unref(msg);
    }

    gst_object_unref(bus);
    gst_element_set_state(pipeline, GST_STATE_NULL);
out_unref:
    gst_object_unref(pipeline);
out_return:
    return 0;
}

四、分析

4.1 创建元素

/* Initialize GStreamer */
gst_init(&argc, &argv);

/* Create the elements */
source = gst_element_factory_make("filesrc", "file-source");
demuxer = gst_element_factory_make("qtdemux", "demuxer");
queue = gst_element_factory_make("queue", "queue");
parser = gst_element_factory_make("h264parse", "parser");
decoder = gst_element_factory_make("mppvideodec", "decoder");
converter = gst_element_factory_make("videoconvert", "converter");
sink = gst_element_factory_make("autovideosink", "sink");

这是我们之前提到的初始化，并创建元素。

/* Create the empty pipeline */
pipeline = gst_pipeline_new("test-pipeline");

/* Check creation */
if (!pipeline || !source || !demuxer || !queue || !parser || !decoder || !converter || !sink) {
    g_printerr("Not all elements could be created.\n");
    return -1;
}

/* Set the location of the input file */
g_object_set(G_OBJECT(source), "location", "13850_h264.mp4", NULL);

/* Build the pipeline */
gst_bin_add_many(GST_BIN(pipeline), source, demuxer, queue, parser, decoder, converter, sink, NULL);

现在我们新建管线，并确将所有元素加入到bin中，方便后面连接。

4.2 Signals

/* Link */
if (!gst_element_link(source, demuxer)) {
    g_printerr("Source and demuxer could not be linked.\n");
    gst_object_unref(pipeline);
    return -1;
}
if (!gst_element_link(queue, parser)) {
    g_printerr("Queue and parser could not be linked.\n");
    gst_object_unref(pipeline);
    return -1;
}
if (!gst_element_link(parser, decoder)) {
    g_printerr("Parser and decoder could not be linked.\n");
    gst_object_unref(pipeline);
    return -1;
}
if (!gst_element_link(decoder, converter)) {
    g_printerr("Decoder and converter could not be linked.\n");
    gst_object_unref(pipeline);
    return -1;
}

/* Manually connect the demuxer to the queue */
g_signal_connect(demuxer, "pad-added", G_CALLBACK(linkElements), queue);

/* Link */
if (!gst_element_link(converter, sink)) {
    g_printerr("Converter and sink could not be linked.\n");
    gst_object_unref(pipeline);
    return -1;
}

在这里，如果使用gst_element_link直接连接demuxer和queue是不行的。此时应当先创建demuxer的pad，然后再与queue相连接。

GSignals是GStreamer中的关键点。当兴趣点出现时，它们允许您收到通知（通过回调的方式）。信号由名称标识，每个 GObject都有自己的信号。

在上述代码中，我们使用g_signal_connect来为demuxer创建一个source pad，并把这个source pad和queue的sink pad相连接。g_signal_connect并不直接处理我们传入的参数，而是将其转发给回调函数。

g_signal_connect(demuxer, "pad-added", G_CALLBACK(linkElements), queue);

static void linkElements(GstElement* element, GstPad* sourcePad, gpointer sinkElement)
{
    GstPad* sinkPad = gst_element_get_static_pad((GstElement*)sinkElement, "sink");
    gst_pad_link(sourcePad, sinkPad);
    gst_object_unref(sinkPad);
}

4.3 GStreamer状态

NULL，元素的null或初始状态。
READY，元素准备好进入暂停状态。
PAUSED，该元素已暂停，它已准备好接受和处理数据。然而，接收器元素只接受一个缓冲区然后阻塞。
PLAYING，播放，时钟运行、数据流动。

这几个状态它们只能从一个状态移动到相邻的状态。例如，NULL不能直接移动到PLAYING状态，而必须先移动到READY、PAUSED状态后，才能进入PLAYING状态。大多数程序只需要关心从PLAYING开始播放，到PAUSED暂停，最后到NULL释放资源。

五、总结

在本篇文章中，我们了解了：

如何使用GSignals通知事件。
如何直接连接GstPads而不是它们的父元素。
GStreamer元素的各种状态。