Gstreamer应用开发手册14：替换管道元件

Bertha 。 2022-12-26 10:09 99阅读 0赞

## [https://zongxp.blog.csdn.net/article/details/105843651][https_zongxp.blog.csdn.net_article_details_105843651] ##

# #

## 动态改变管道 ##

在本节中，我们讨论一些用于动态修改管道的技术。我们正在专门讨论在PLAYING状态下且不中断数据流的情况下更改管道。

建立动态管道时，需要考虑一些重要事项：

*  从管道中删除元件时，请确保未链接的pad上没有数据流，因为这将导致致命的管道错误。断开pad连接之前，请始终阻塞source pad（在推模式下）或sink pad（在拉模式下）。
 *  将元件添加到管道中时，请确保在允许数据流之前，将元件置于正确的状态，通常与父级处于同一状态。新创建元件时，该元件处于 NULL状态，并且在接收数据时将返回错误。
 *  将元件添加到管道时，GStreamer默认情况下会将元件的时钟和基准时间设置为管道的当前值。这意味着该元件将能够构造与管道中其他元件相同的管道运行时间。这意味着接收器将像管道中的其他接收器一样同步缓冲区，并且源将生成运行时间与其他源匹配的缓冲区。
 *  当从上游链取消链接元件时，请始终确保通过向元件接收器发送一个EOS事件并等待EOS元件离开（带有事件探测器）来刷新元件中所有排队的数据。如果不执行刷新，则将丢失由未链接的元件缓冲的数据。这可能会导致简单的帧丢失（一些视频帧，几毫秒的音频等），但是如果您删除多路复用器（在某些情况下为编码器或类似元件），则可能会损坏文件，由于某些相关的元数据（标题，查找/索引表，内部同步标签）可能未正确存储或更新，因此无法正确播放。
 *  实时源将产生与管道的running-time相等的缓冲区running-time。没有活动源的管道会产生running-time从0开始的缓冲区。 同样，在进行刷新查找之后，这些管道会将其重置running-time为0。该running-time可被改变gst\_pad\_set\_offset ()。重要的是要知道running-time管道中的元件，以保持同步。
 *  添加元件可能会更改管道的状态。例如，添加未预卷的接收器会使管道返回到预卷状态。例如，删除未预卷的接收器可能会将管道更改为PAUSED和PLAYING状态。添加实时源会取消预滚动阶段，并使管道处于播放状态。添加任何活动元件也可能会更改管道的延迟。添加或删除管道的元件可能会更改管道的时钟选择。如果新添加的元件提供了时钟，则管道使用新时钟可能会更好。另一方面，如果删除了为流水线提供时钟的元件，则必须选择新的时钟。
 *  添加和删除元件可能会导致上游或下游元件重新协商上限和/或分配器。您实际上不需要从应用程序执行任何操作，插件可以在很大程度上适应新的管道拓扑，从而优化其格式和分配策略。重要的是，当您添加，删除或更改管道中的元件时，管道可能需要协商新格式，这可能会失败。通常，您可以通过在需要的地方插入正确的转换器元件来解决此问题。另请参见更改管道中的元件。

GStreamer提供了对几乎所有动态管道修改的支持，但是您需要了解一些详细信息，然后才能执行此操作而不会引起管道错误

在以下各节中，我们将演示一些典型的修改用例。

### 更改管道中的元件 ###

在此示例中，我们具有以下元件链：

1.  `- ----. .----------. .---- -`
2.  `element1 | | element2 | | element3`
3.  `src -> sink src -> sink`
4.  `- ----' '----------' '---- -`

我们希望在管道处于PLAYING状态时用element4替换element2。假设element2是可视化，并且您想在管道中切换可视化。

我们不能仅仅将element2的sink pad与element1的src pad断开链接，因为这将使element1的src pad保持未链接状态，并在将数据推入src pad时导致流水错误。应该在将element2替换为element4之前，阻止来自element1的src pad的数据流，然后恢复数据流，如以下步骤所示：

*  用阻塞垫探针阻塞element1的src pad。当pad被阻塞时，探测回调将被调用。
 *  在block回调内部，element1和element2之间没有任何流通，直到被阻塞为止，没有任何流通。
 *  取消链接element1和element2。
 *  确保数据已从element2中清除。有些元件可能会在内部保留一些数据，因此您需要通过推EOS到element2确保不会丢失任何数据，如下所示：将事件探针放在element2的src pad上。发送EOS到element2的sink pad；这样可以确保元件2中的所有数据都被强制删除；等待EOS事件出现在element2的src pad上。当EOS收到，拖放和删除事件探测器。
 *  取消链接element2和element3。现在，您还可以从管道中删除element2并将状态设置为NULL。
 *  如果尚未添加element4，则将其添加到管道中。链接element4和element3，链接element1和element4。
 *  确保element4与管道中其余元件的状态相同。它至少应处于PAUSED可接收缓冲区和事件的状态。
 *  取消阻止element1的src pad,这将使新数据进入element4并继续流式传输。

当source pad被阻塞时，即管道中有数据流时，上述算法将起作用。如果没有数据流，那么更改元件也没有任何意义，因此该算法也可以在PAUSED状态中使用。

本示例每秒更改一次简单管道上的视频效果：

\#include <gst/gst.h>  
   
static gchar \*opt\_effects = NULL;  
   
\#define DEFAULT\_EFFECTS "identity,exclusion,navigationtest," \\  
    "agingtv,videoflip,vertigotv,gaussianblur,shagadelictv,edgetv"  
   
static GstPad \*blockpad;  
static GstElement \*conv\_before;  
static GstElement \*conv\_after;  
static GstElement \*cur\_effect;  
static GstElement \*pipeline;  
   
static GQueue effects = G\_QUEUE\_INIT;  
   
static GstPadProbeReturn  
event\_probe\_cb (GstPad \* pad, GstPadProbeInfo \* info, gpointer user\_data)  
\{  
  GMainLoop \*loop = user\_data;  
  GstElement \*next;  
   
  if (GST\_EVENT\_TYPE (GST\_PAD\_PROBE\_INFO\_DATA (info)) != GST\_EVENT\_EOS)  
    return GST\_PAD\_PROBE\_PASS;  
   
  gst\_pad\_remove\_probe (pad, GST\_PAD\_PROBE\_INFO\_ID (info));  
   
  /\* push current effect back into the queue \*/  
  g\_queue\_push\_tail (&effects, gst\_object\_ref (cur\_effect));  
  /\* take next effect from the queue \*/  
  next = g\_queue\_pop\_head (&effects);  
  if (next == NULL) \{  
    GST\_DEBUG\_OBJECT (pad, "no more effects");  
    g\_main\_loop\_quit (loop);  
    return GST\_PAD\_PROBE\_DROP;  
  \}  
   
  g\_print ("Switching from '%s' to '%s'..\\n", GST\_OBJECT\_NAME (cur\_effect),  
      GST\_OBJECT\_NAME (next));  
   
  gst\_element\_set\_state (cur\_effect, GST\_STATE\_NULL);  
   
  /\* remove unlinks automatically \*/  
  GST\_DEBUG\_OBJECT (pipeline, "removing %" GST\_PTR\_FORMAT, cur\_effect);  
  gst\_bin\_remove (GST\_BIN (pipeline), cur\_effect);  
   
  GST\_DEBUG\_OBJECT (pipeline, "adding   %" GST\_PTR\_FORMAT, next);  
  gst\_bin\_add (GST\_BIN (pipeline), next);  
   
  GST\_DEBUG\_OBJECT (pipeline, "linking..");  
  gst\_element\_link\_many (conv\_before, next, conv\_after, NULL);  
   
  gst\_element\_set\_state (next, GST\_STATE\_PLAYING);  
   
  cur\_effect = next;  
  GST\_DEBUG\_OBJECT (pipeline, "done");  
   
  return GST\_PAD\_PROBE\_DROP;  
\}  
   
static GstPadProbeReturn  
pad\_probe\_cb (GstPad \* pad, GstPadProbeInfo \* info, gpointer user\_data)  
\{  
  GstPad \*srcpad, \*sinkpad;  
   
  GST\_DEBUG\_OBJECT (pad, "pad is blocked now");  
   
  /\* remove the probe first \*/  
  gst\_pad\_remove\_probe (pad, GST\_PAD\_PROBE\_INFO\_ID (info));  
   
  /\* install new probe for EOS \*/  
  srcpad = gst\_element\_get\_static\_pad (cur\_effect, "src");  
  gst\_pad\_add\_probe (srcpad, GST\_PAD\_PROBE\_TYPE\_BLOCK |  
      GST\_PAD\_PROBE\_TYPE\_EVENT\_DOWNSTREAM, event\_probe\_cb, user\_data, NULL);  
  gst\_object\_unref (srcpad);  
   
  /\* push EOS into the element, the probe will be fired when the  
   \* EOS leaves the effect and it has thus drained all of its data \*/  
  sinkpad = gst\_element\_get\_static\_pad (cur\_effect, "sink");  
  gst\_pad\_send\_event (sinkpad, gst\_event\_new\_eos ());  
  gst\_object\_unref (sinkpad);  
   
  return GST\_PAD\_PROBE\_OK;  
\}  
   
static gboolean  
timeout\_cb (gpointer user\_data)  
\{  
  gst\_pad\_add\_probe (blockpad, GST\_PAD\_PROBE\_TYPE\_BLOCK\_DOWNSTREAM,  
      pad\_probe\_cb, user\_data, NULL);  
   
  return TRUE;  
\}  
   
static gboolean  
bus\_cb (GstBus \* bus, GstMessage \* msg, gpointer user\_data)  
\{  
  GMainLoop \*loop = user\_data;  
   
  switch (GST\_MESSAGE\_TYPE (msg)) \{  
    case GST\_MESSAGE\_ERROR:\{  
      GError \*err = NULL;  
      gchar \*dbg;  
   
      gst\_message\_parse\_error (msg, &err, &dbg);  
      gst\_object\_default\_error (msg->src, err, dbg);  
      g\_clear\_error (&err);  
      g\_free (dbg);  
      g\_main\_loop\_quit (loop);  
      break;  
    \}  
    default:  
      break;  
  \}  
  return TRUE;  
\}  
   
int  
main (int argc, char \*\*argv)  
\{  
  GOptionEntry options\[\] = \{  
    \{"effects", 'e', 0, G\_OPTION\_ARG\_STRING, &opt\_effects,  
        "Effects to use (comma-separated list of element names)", NULL\},  
    \{NULL\}  
  \};  
  GOptionContext \*ctx;  
  GError \*err = NULL;  
  GMainLoop \*loop;  
  GstElement \*src, \*q1, \*q2, \*effect, \*filter1, \*filter2, \*sink;  
  gchar \*\*effect\_names, \*\*e;  
   
  ctx = g\_option\_context\_new ("");  
  g\_option\_context\_add\_main\_entries (ctx, options, NULL);  
  g\_option\_context\_add\_group (ctx, gst\_init\_get\_option\_group ());  
  if (!g\_option\_context\_parse (ctx, &argc, &argv, &err)) \{  
    g\_print ("Error initializing: %s\\n", err->message);  
    g\_clear\_error (&err);  
    g\_option\_context\_free (ctx);  
    return 1;  
  \}  
  g\_option\_context\_free (ctx);  
   
  if (opt\_effects != NULL)  
    effect\_names = g\_strsplit (opt\_effects, ",", -1);  
  else  
    effect\_names = g\_strsplit (DEFAULT\_EFFECTS, ",", -1);  
   
  for (e = effect\_names; e != NULL && \*e != NULL; ++e) \{  
    GstElement \*el;  
   
    el = gst\_element\_factory\_make (\*e, NULL);  
    if (el) \{  
      g\_print ("Adding effect '%s'\\n", \*e);  
      g\_queue\_push\_tail (&effects, el);  
    \}  
  \}  
   
  pipeline = gst\_pipeline\_new ("pipeline");  
   
  src = gst\_element\_factory\_make ("videotestsrc", NULL);  
  g\_object\_set (src, "is-live", TRUE, NULL);  
   
  filter1 = gst\_element\_factory\_make ("capsfilter", NULL);  
  gst\_util\_set\_object\_arg (G\_OBJECT (filter1), "caps",  
      "video/x-raw, width=320, height=240, "  
      "format=\{ I420, YV12, YUY2, UYVY, AYUV, Y41B, Y42B, "  
      "YVYU, Y444, v210, v216, NV12, NV21, UYVP, A420, YUV9, YVU9, IYU1 \}");  
   
  q1 = gst\_element\_factory\_make ("queue", NULL);  
   
  blockpad = gst\_element\_get\_static\_pad (q1, "src");  
   
  conv\_before = gst\_element\_factory\_make ("videoconvert", NULL);  
   
  effect = g\_queue\_pop\_head (&effects);  
  cur\_effect = effect;  
   
  conv\_after = gst\_element\_factory\_make ("videoconvert", NULL);  
   
  q2 = gst\_element\_factory\_make ("queue", NULL);  
   
  filter2 = gst\_element\_factory\_make ("capsfilter", NULL);  
  gst\_util\_set\_object\_arg (G\_OBJECT (filter2), "caps",  
      "video/x-raw, width=320, height=240, "  
      "format=\{ RGBx, BGRx, xRGB, xBGR, RGBA, BGRA, ARGB, ABGR, RGB, BGR \}");  
   
  sink = gst\_element\_factory\_make ("ximagesink", NULL);  
   
  gst\_bin\_add\_many (GST\_BIN (pipeline), src, filter1, q1, conv\_before, effect,  
      conv\_after, q2, sink, NULL);  
   
  gst\_element\_link\_many (src, filter1, q1, conv\_before, effect, conv\_after,  
      q2, sink, NULL);  
   
  gst\_element\_set\_state (pipeline, GST\_STATE\_PLAYING);  
   
  loop = g\_main\_loop\_new (NULL, FALSE);  
   
  gst\_bus\_add\_watch (GST\_ELEMENT\_BUS (pipeline), bus\_cb, loop);  
   
  g\_timeout\_add\_seconds (1, timeout\_cb, loop);  
   
  g\_main\_loop\_run (loop);  
   
  gst\_element\_set\_state (pipeline, GST\_STATE\_NULL);  
  gst\_object\_unref (pipeline);  
   
  return 0;  
\}

注意我们如何在videoconvert效果之前和之后添加元件。这是必需的，因为某些元件可能在不同的色彩空间中运行；通过插入转换元件，我们可以帮助确保可以协商适当的格式

[https_zongxp.blog.csdn.net_article_details_105843651]: https://zongxp.blog.csdn.net/article/details/105843651