Gstreamer 动态更改源元素

好的，所以我相信我已经找到了答案，我将在这里发布这个，以拯救任何偶然发现这个问题的人。

答案是创建一对垫探针来处理管道中的数据清除。 我通过创建两个 pad 探测器回调来实现这一点：一个用于捕获管道以开始刷新过程，另一个用于在管道刷新后处理 rtspsrc 元素的重新创建。 第一个焊盘探头可以放在任何地方，所以我把它放在我的depay元件上。 第二个焊盘探头必须位于最后一个数据处理元素的源上。 所以不是最终的接收器元素。 对于上面的管道，这是"vpe"元素。

为此，我将流结束 (EOS( 信号传递给 depay 元素，然后在 vpe 元素的 src pad 处进行垫探测回调，以在 EOS 退出 VPE 时捕获它。 如果EOS到达waylandsink，管道将简单地关闭，你必须重新启动整个事情。

vpe = gst_bin_get_by_name(GST_BIN(data), "vpe");
srcPad = gst_element_get_static_pad(vpe, "src");
gst_pad_add_probe(srcPad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM, event_probe, data, NULL);
//push EOS into the element, wait for the EOS to appear on the srcpad
depay = gst_bin_get_by_name(GST_BIN(data), "depay");
sinkPad = gst_element_get_static_pad(depay, "sink");
gst_pad_send_event(sinkPad, gst_event_new_eos());    
return GST_PAD_PROBE_OK;

以及处理该EOS的回调：

static GstPadProbeReturn event_probe(GstPad *pad, GstPadProbeInfo *info, gpointer data){
GstElement *rtspsrcOld, *rtspsrcNew, *depay;
qDebug("event_probe");
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));
rtspsrcOld = gst_bin_get_by_name(GST_BIN(data), "rtspsrc");
if(rtspsrcOld){
qDebug("found rtspsrcOld");
depay = gst_bin_get_by_name(GST_BIN(data), "depay");
gst_element_unlink(rtspsrcOld, depay);
gst_bin_remove(GST_BIN(data), rtspsrcOld); //remove old rtspsrc from pipeline, should unlink from depay automatically.
rtspsrcNew = gst_element_factory_make("rtspsrc", "rtspsrc");
g_object_set(rtspsrcNew, "location", NEW_URI, "latency", 0, NULL);
g_signal_connect(G_OBJECT(rtspsrcNew), "pad-added", G_CALLBACK(on_rtsp_pad_added), data);
gst_bin_add(GST_BIN(data), rtspsrcNew);
gst_element_set_state(GST_ELEMENT(data), GST_STATE_PLAYING);
return GST_PAD_PROBE_DROP;
}
return GST_PAD_PROBE_DROP;
}

我试图做同样的事情。我刚开始使用gstreamer。在理解了T. Wallis的意思之后，我想用一个简单的管道来测试它。不幸的是，新的 rtspsrc 元素与管道的最终链接不起作用。但是，我认为错误在其他地方。我将再次浏览代码并阅读 gstreamer 的动态管道操作过程。但我不确定，我是否能够这么快找到错误。这是我的代码(这是我第一次在堆栈溢出上发帖，对不起潜在的nogos(：

#include <gst/gst.h>
#include <gst/gstpad.h>
#include <gst/rtsp/gstrtsp.h>
#include <unistd.h>
#include <time.h>
#include <stdbool.h>
typedef struct _CustomData {
GstElement *streaming_pipe;
GstElement *src;
GstElement *depay;
GstElement *decoder;
GstElement *sink;
GMainLoop *m_loop; 
gboolean change_url;
gboolean url1;
clock_t startT;
} CustomData;
static void on_rtsp_pad_added(GstElement *element, GstPad *new_pad,  CustomData *data){
gchar *name;
GstCaps *caps;

caps = gst_caps_from_string("application/x-rtp");
name = gst_pad_get_name(new_pad);

if(!gst_element_link_pads_filtered(element, name, data->depay, "sink", caps)){
g_print("npad_added: failed to link elements"); //ERROR when linking the new rtspsrc after breaking up the pipeline
}
g_free(name);
data->startT = clock();
}
static GstPadProbeReturn event_probe(GstPad *pad, GstPadProbeInfo *info, CustomData *data){
GstElement *rtspsrcOld, *rtspsrcNew, *depay;
if(GST_EVENT_TYPE(GST_PAD_PROBE_INFO_DATA(info)) != GST_EVENT_EOS){
g_print("n Not an EOS event; pass probe return");
return GST_PAD_PROBE_PASS;
}
gst_pad_remove_probe(pad, GST_PAD_PROBE_INFO_ID(info));
rtspsrcOld = gst_bin_get_by_name(GST_BIN(data->streaming_pipe), "rtspsrc");
if(rtspsrcOld){
depay = gst_bin_get_by_name(GST_BIN(data->streaming_pipe), "depay");
gst_element_unlink(rtspsrcOld, depay);
gst_bin_remove(GST_BIN(data->streaming_pipe), rtspsrcOld); //remove old rtspsrc from pipeline, should unlink from depay automatically.
rtspsrcNew = gst_element_factory_make("rtspsrc", "rtspsrc123");
g_object_set(rtspsrcNew, "location", "rtsp://xxxx/axis-media/media.amp?videocodec=h264&resolution=480x270", "latency", 0, NULL);
g_signal_connect(rtspsrcNew, "pad-added", G_CALLBACK(on_rtsp_pad_added), data);
gst_bin_add(GST_BIN(data->streaming_pipe), rtspsrcNew);
gst_element_set_state(GST_ELEMENT(data->streaming_pipe), GST_STATE_PLAYING);
g_print("n set playingn");
return GST_PAD_PROBE_DROP;
}
return GST_PAD_PROBE_DROP;
}

static GstPadProbeReturn cb_have_data (GstPad *pad, GstPadProbeInfo *info, CustomData *data) {
g_print("nPROBE CALLBACK!");
g_print("Time: %f", ((double) (clock() - data->startT)) / CLOCKS_PER_SEC);
if(((double) (clock() - data->startT)) / CLOCKS_PER_SEC > 0.04){
data->change_url = true;
data->startT = clock();
}
if(data->change_url){
g_print("nIF PROBE CALLBACK!");
GstPad *srcPad, *sinkPad;
srcPad = gst_element_get_static_pad(data->decoder, "src");
gst_pad_add_probe(srcPad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM, event_probe, data, NULL);

//push EOS into the element, wait for the EOS to appear on the srcpad
sinkPad = gst_element_get_static_pad(data->depay, "sink");
gst_pad_send_event(sinkPad, gst_event_new_eos());    
data->change_url = false;
}
return GST_PAD_PROBE_OK;
}

int main(int argc, char *argv[])
{
/* Initialize GStreamer */
gst_init (&argc, &argv);
CustomData data;
GstStateChangeReturn ret;
GstPad *pad;
data.m_loop = g_main_loop_new(NULL, FALSE);
//create pipeline elements
data.streaming_pipe = gst_pipeline_new("display_pipeline");
data.src = gst_element_factory_make("rtspsrc", "rtspsrc");
data.depay = gst_element_factory_make("rtph264depay", "depay");
data.decoder = gst_element_factory_make("avdec_h264", "decoder");
data.sink = gst_element_factory_make("autovideosink", NULL);
data.change_url = false;
data.url1 = false;
if(!(data.streaming_pipe || data.src || data.depay || data.decoder || data.sink)){
g_print("could not create pipeline elements");
exit(1);
}
g_object_set(G_OBJECT(data.src), "location", "rtsp://xxxx/axis-media/media.amp", "latency", 0, NULL);
g_signal_connect(data.src, "pad-added", G_CALLBACK(on_rtsp_pad_added), &data);
//add and link elements to create full pipeline
gst_bin_add_many(GST_BIN(data.streaming_pipe), data.src, data.depay, data.decoder,  data.sink, NULL);
if(!gst_element_link_many(data.depay, data.decoder,  data.sink, NULL)){
g_print("cannot link elements"); 
exit(1);
}
pad = gst_element_get_static_pad (data.depay, "src");
if(pad == NULL){
g_print("COULD NOT GET STATIC PAD");
}
gst_pad_add_probe (pad, GST_PAD_PROBE_TYPE_BUFFER,
(GstPadProbeCallback) cb_have_data, &data, NULL);
gst_object_unref (pad);
ret = gst_element_set_state (data.streaming_pipe, 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.streaming_pipe);
return -1;
}

g_main_loop_run (data.m_loop);
}