GStreamer 基础教程三: 动态管道

目标

我们可以在应用程序开始时定义整体管道，也可以在有足够信息时“动态”构建管道。

这篇教程的目标是搞清楚如下问题：

• 如何在链接元件时获得更精细的控制？
• 如何收到感兴趣事件的通知，并及时反应？
• 元件可能处于的哪些不同的状态？

介绍

这篇教程中的管道在设置为 “Playing” 状态之前并未完全构建好。
这没问题。如果我们不采取进一步的行动，数据将到达管道的尽头，管道将会产生一个错误消息并停止。但这里我们将会采取点行动。

在这个例子中，我们将打开一个包含多路媒体的文件(多路复用muxed), 也就是将音频和视频存储在一个容器文件中。

负责打开这样的容器的元件称为分离器(demuxer), 常见的容器格式有 Matroska(MKV), Quick Time(QT, MOV), Ogg 或 Advanced Systems Format(ASF, WMV, WMA).

如果一个容器嵌入了多路媒体流 (例如一路视频和两路音频)，分离器 demuxer 会分开它们，并通过不同的输出端口发布出去。这样就可以在管道中创建不同的分支，处理不同类型的数据。

GStreamer 元件相互通信所通过的端口称为 Pad(GstPad)，有 Sink Pad(数据通过其进入元件) 和 Src Pad(数据由其离开元件)。自然地，Src Element 仅包含 Src Pad, Sink Element 仅包含 Sink Pad, 而 Filter element 既包含 src pad 也包含 sink pad.

source element filter element sink element

一个分离器 demuxer 包含一个 sink pad （聚合的数据通过它到达), 和多个 source pads (对应于在容器中找到的媒体流)

为了完整起见，这里有一个简化的管道，其中包含一个分离器和两个分支，一个用于音频，一个用于视频。 这不是本示例中将构建的管道：

处理分离器时的主要复杂性在于，它们无法产生任何信息，直到它们收到一些数据并且有机会查看容器以了解里面的内容。

也就是说，分离器开始时没有其他元件可以链接的 source pad，因此管道必须在分离器处终止。

解决方案是构建从源元件到分离器的管道，并将其设置为运行 (playing 状态)。 当分离器收到足够的信息以了解容器中流的数量和类型时，它将开始创建 source pad。 现在是我们完成管道构建并将其连接到新添加的 demux source pad 的最佳时机。

为简单起见，在本例中，我们将仅链接到音频 pad 并忽略视频流。

示例

大致流程如下

@startuml

start
: 初始化 gst_init();
: 创建元件 gst_element_factory_make();
: 创建管道 gst_pipeline_new();
: 添加元件到管道 gst_bin_add_many();
: 将元件连接起来 gst_element_link 
  (除了 source element);
: 设置元件属性 g_object_set();
: 设置 source 元件的信号 “pad-added” 的回调;
: 设置管道状态 gst_element_set_state();
: 等待信号发生 gst_signal_emit(由 demuxer 触发)
 等待总线消息 gst_bus_timed_pop_filtered();

fork
  :pad-added signal triggered;
  :pad_added_handler;
fork again
  :gst_bus_timed_pop_filtered
  -> GST_MESSAGE_ERROR;
fork again
  :gst_bus_timed_pop_filtered
  -> GST_MESSAGE_EOS;
fork again
  :gst_bus_timed_pop_filtered
  -> GST_MESSAGE_ERROR;
fork again
  :gst_bus_timed_pop_filtered
  -> GST_MESSAGE_STATE_CHANGED;
end merge
stop

@enduml

源代码

#include <gst/gst.h>

/* Structure to contain all our information, so we can pass it to callbacks */
typedef struct _CustomData {
  GstElement *pipeline;
  GstElement *source;
  GstElement *convert;
  GstElement *resample;
  GstElement *sink;
} CustomData;

/* Handler for the pad-added signal */
static void pad_added_handler (GstElement *src, GstPad *pad, CustomData *data);

int main(int argc, char *argv[]) {
  CustomData data;
  GstBus *bus;
  GstMessage *msg;
  GstStateChangeReturn ret;
  gboolean terminate = FALSE;

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

  /* Create the elements */
  data.source = gst_element_factory_make ("uridecodebin", "source");
  data.convert = gst_element_factory_make ("audioconvert", "convert");
  data.resample = gst_element_factory_make ("audioresample", "resample");
  data.sink = gst_element_factory_make ("autoaudiosink", "sink");

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

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

  /* Build the pipeline. Note that we are NOT linking the source at this
   * point. We will do it later. */
  gst_bin_add_many (GST_BIN (data.pipeline), data.source, data.convert, data.resample, data.sink, NULL);
  if (!gst_element_link_many (data.convert, data.resample, data.sink, NULL)) {
    g_printerr ("Elements could not be linked.\n");
    gst_object_unref (data.pipeline);
    return -1;
  }

  /* Set the URI to play */
  g_object_set (data.source, "uri", "https://gstreamer.freedesktop.org/data/media/sintel_trailer-480p.webm", NULL);

  /* Connect to the pad-added signal */
  g_signal_connect (data.source, "pad-added", G_CALLBACK (pad_added_handler), &data);

  /* Start playing */
  ret = gst_element_set_state (data.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 (data.pipeline);
    return -1;
  }

  /* Listen to the bus */
  bus = gst_element_get_bus (data.pipeline);
  do {
    msg = gst_bus_timed_pop_filtered (bus, GST_CLOCK_TIME_NONE,
        GST_MESSAGE_STATE_CHANGED | 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);
          terminate = TRUE;
          break;
        case GST_MESSAGE_EOS:
          g_print ("End-Of-Stream reached.\n");
          terminate = TRUE;
          break;
        case GST_MESSAGE_STATE_CHANGED:
          /* We are only interested in state-changed messages from the pipeline */
          if (GST_MESSAGE_SRC (msg) == GST_OBJECT (data.pipeline)) {
            GstState old_state, new_state, pending_state;
            gst_message_parse_state_changed (msg, &old_state, &new_state, &pending_state);
            g_print ("Pipeline state changed from %s to %s:\n",
                gst_element_state_get_name (old_state), gst_element_state_get_name (new_state));
          }
          break;
        default:
          /* We should not reach here */
          g_printerr ("Unexpected message received.\n");
          break;
      }
      gst_message_unref (msg);
    }
  } while (!terminate);

  /* Free resources */
  gst_object_unref (bus);
  gst_element_set_state (data.pipeline, GST_STATE_NULL);
  gst_object_unref (data.pipeline);
  return 0;
}

/* This function will be called by the pad-added signal */
static void pad_added_handler (GstElement *src, GstPad *new_pad, CustomData *data) {
  GstPad *sink_pad = gst_element_get_static_pad (data->convert, "sink");
  GstPadLinkReturn ret;
  GstCaps *new_pad_caps = NULL;
  GstStructure *new_pad_struct = NULL;
  const gchar *new_pad_type = NULL;

  g_print ("Received new pad '%s' from '%s':\n", GST_PAD_NAME (new_pad), GST_ELEMENT_NAME (src));

  /* If our converter is already linked, we have nothing to do here */
  if (gst_pad_is_linked (sink_pad)) {
    g_print ("We are already linked. Ignoring.\n");
    goto exit;
  }

  /* Check the new pad's type */
  new_pad_caps = gst_pad_get_current_caps (new_pad);
  new_pad_struct = gst_caps_get_structure (new_pad_caps, 0);
  new_pad_type = gst_structure_get_name (new_pad_struct);
  if (!g_str_has_prefix (new_pad_type, "audio/x-raw")) {
    g_print ("It has type '%s' which is not raw audio. Ignoring.\n", new_pad_type);
    goto exit;
  }

  /* Attempt the link */
  ret = gst_pad_link (new_pad, sink_pad);
  if (GST_PAD_LINK_FAILED (ret)) {
    g_print ("Type is '%s' but link failed.\n", new_pad_type);
  } else {
    g_print ("Link succeeded (type '%s').\n", new_pad_type);
  }

exit:
  /* Unreference the new pad's caps, if we got them */
  if (new_pad_caps != NULL)
    gst_caps_unref (new_pad_caps);

  /* Unreference the sink pad */
  gst_object_unref (sink_pad);
}

上述代码可通过如下命令行编译

gcc basic-tutorial-3.c -o basic-tutorial-3 `pkg-config --cflags --libs gstreamer-1.0`

关键代码解析

其中有意思的就是 pad-added 的回调函数 pad_added_handler，它会将新创建出来的 newpad(urldecodebin 新建出来的) 与 sink_pad( audioconvert 的) 连接起来。

  /* Connect to the pad-added signal */
  g_signal_connect (data.source, "pad-added", G_CALLBACK (pad_added_handler),
      &data);

/* This function will be called by the pad-added signal */
static void
pad_added_handler (GstElement * src, GstPad * new_pad, CustomData * data)
{
  /* Attempt the link */
  ret = gst_pad_link (new_pad, sink_pad);
  if (GST_PAD_LINK_FAILED (ret)) {
    g_print ("Type is '%s' but link failed.\n", new_pad_type);
  } else {
    g_print ("Link succeeded (type '%s').\n", new_pad_type);
  }
}

原来的 pipeline 是

audioconvert -> audioresample -> autoaudiosink

当 uridecodebin 的 src_pad 按需创建出来后，将 uridecodebin 的 src_pad 与 audioconvert 的 sink_pad 连接起来，变成

urldecodebin -> audioconvert -> audioresample -> autoaudiosink

结论

这样我们了解到了如下知识

• 如何使用 GSignals 收到事件通知

  • 使用 g_signal_connect

• 如何直接连接 GstPad 而不是其父元素

  • 使用 gst_pad_link

• GStreamer 元素的各种状态:

  • 使用 gst_element_set_state 设置状态，可以通过观察 GST_MESSAGE_STATE_CHANGED 类型的消息来观察管道的状态变化

    • GST_STATE_NULL：已停用，元件不占用资源
    • GST_STATE_READY：检查并分配资源
    • GST_STATE_PAUSED：预滚动，即为每个接收器 sink 获取一个缓冲区
    • GST_STATE_PLAYING：活动数据流，运行时间在不断增加

还有如何组合了这些项目来构建动态管道，该管道不是在程序启动时定义的，而是在有关媒体的信息可用时创建的。