文件接收器中的 Gstreamer EOS 消息处理以动态更改位置

Question

尝试即时切换输出文件，但无法处理 EOS。

http://gstreamer-devel.966125.n4.nabble.com/Dynamically-updating-filesink-location-at-run-time-on-the-fly-td4660569.html

引用:

Assuming you have a pipeline that looks like this:
audiosrc --> encoder --> mux --> filesink

then you'll need to change it to:
audiosrc --> encoder --> queue --> muxsink_bin
where muxsink_bin is a bin
ghostpad --> mux --> filesink

then the procedure is:
1 - Block the queue srcpad using gst_pad_set_blocked_async()
2 - In the blocked callback:
2a - unlink muxsink_bin with gst_pad_unlink()
2b - send an EOS event to the muxsink_bin sink pad with gst_pad_send_event()
2b - create a new muxsink_bin
2c - set filesink location
2d - add the new bin to the pipeline with gst_bin_add()
2e - sync with parent using gst_element_sync_state_with_parent()
2f - link it to the queue srcpad with gst_pad_link()
2g - unblock the queue srcpad with gst_pad_set_blocked_async(). When the unblocked callback occurs you're recording again & no data has been lost. No action is required in the unblocked callback

3 - handle the EOS & delete the old muxsink_bin. I had a msg handler that I installed in my bin_init() function using "gstbin_class->handle_message = GST_DEBUG_FUNCPTR(msg_handler)" & in the handler:
3a - lock the bin state with gst_element_set_locked_state()
3b - set the state to NULL with gst_element_set_state()
3c - remove it from the pipeline with gst_bin_remove()

That's it. The only thing to be mindful of is that data must be flowing thru the pipeline for this to work.

Paddy

除了旧流水线的完成之外，主序列正常工作。

困难在于第 3 点:我可以将 EOS 发送到 ghostpad，然后 filesink 得到它。但是如何捕获那个EOS呢？

“使用 gstbin_class->handle_message = GST_DEBUG_FUNCPTR(msg_handler) 安装消息处理程序”是什么意思？

Answer 1

有消息转发。

必须在总线上启用:

g_object_set(G_OBJECT(bin), "message-forward", TRUE, 0);

处理:

case GST_MESSAGE_ELEMENT:
{
    const GstStructure *s = gst_message_get_structure (msg);

    if (gst_structure_has_name (s, "GstBinForwarded"))
    {
        GstMessage *forward_msg = NULL;

        gst_structure_get (s, "message", GST_TYPE_MESSAGE, &forward_msg, NULL);
        if (GST_MESSAGE_TYPE (forward_msg) == GST_MESSAGE_EOS)
        {
            g_print ("EOS from element %s\n",
                    GST_OBJECT_NAME (GST_MESSAGE_SRC (forward_msg)));
            DestroyBin();
            CreateNewBin();
            RemovePad();
        }
        gst_message_unref (forward_msg);
    }
}

完整代码:

#include <gst/gst.h>
#include <iostream>
#include <cstring>
#include <cstdio>
static gchar *opt_effects = NULL;

#define DEFAULT_EFFECTS "identity,exclusion,navigationtest," \
        "agingtv,videoflip,vertigotv,gaussianblur,shagadelictv,edgetv"

static GstElement *pipeline;
static GstElement * muxer;
static GstElement * sink;
static GstElement * q2;
static int i=0;
GstElement * bin;
GstPad * muxerSinkPad;

gulong probeId;

static GQueue effects = G_QUEUE_INIT;

void CreateNewBin();
void DestroyBin();
void ChangeLocation();
void RemovePad();

static GstPadProbeReturn
pad_probe_cb (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
    GstPad *sinkPad = gst_element_get_static_pad(bin, "sink");
    gst_pad_unlink(pad, sinkPad);
    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)
{
    static int i=0;
    if(i==0)
    {
        GstPad * q2SrcPad;
        q2SrcPad = gst_element_get_static_pad(q2, "src");
        std::cout << "Timeout: " << q2SrcPad << std::endl;

        probeId = gst_pad_add_probe (q2SrcPad, GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM,
                pad_probe_cb, user_data, NULL);
        gst_object_unref(q2SrcPad);

        return TRUE;
    }
return FALSE;
}

static gboolean
bus_cb (GstBus * bus, GstMessage * msg, gpointer user_data)
{
    GMainLoop *loop = (GMainLoop*)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_error_free (err);
        g_free (dbg);
        g_main_loop_quit (loop);
        break;
    }
    case GST_EVENT_EOS:
        std::cout << "EOS message is got" << std::endl;
        break;

    case GST_MESSAGE_ELEMENT:
    {
        const GstStructure *s = gst_message_get_structure (msg);

        if (gst_structure_has_name (s, "GstBinForwarded"))
        {
            GstMessage *forward_msg = NULL;

            gst_structure_get (s, "message", GST_TYPE_MESSAGE, &forward_msg, NULL);
            if (GST_MESSAGE_TYPE (forward_msg) == GST_MESSAGE_EOS)
            {
                g_print ("EOS from element %s\n",
                        GST_OBJECT_NAME (GST_MESSAGE_SRC (forward_msg)));
                DestroyBin();
                CreateNewBin();
                RemovePad();
            }
            gst_message_unref (forward_msg);
        }
    }
        break;

    default:
        break;
    }
    return TRUE;
}

int
main (int argc, char **argv)
{

    GError *err = NULL;
    GMainLoop *loop;
    GstElement *src, *q1,/* *q2,*/ /**effect,*/ /**filter1*//*, *filter2*/ *encoder;/*, *sink*/;

    gst_init(&argc, &argv);

    pipeline = gst_pipeline_new ("pipeline");

    src = gst_element_factory_make ("videotestsrc", NULL);

    //Create a caps filter between videosource videoconvert
    std::string capsString = "video/x-raw,format=YV12,width=320,height=240,framerate=30/1";
    GstCaps * dataFilter = gst_caps_from_string(capsString.c_str());

    q1 = gst_element_factory_make ("queue", NULL);

    encoder = gst_element_factory_make ("x264enc", NULL);

    q2 = gst_element_factory_make("queue", NULL);

    gst_bin_add_many(GST_BIN(pipeline), src, q1, encoder, q2, 0);
    gboolean link = gst_element_link_filtered(src, q1, dataFilter);
    link &= gst_element_link(q1, encoder);
    link &= gst_element_link(encoder, q2);

    CreateNewBin();

    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 (10, timeout_cb, loop);

    g_main_loop_run (loop);

    gst_element_set_state (pipeline, GST_STATE_NULL);
    gst_object_unref (pipeline);

    return 0;
}

void RemovePad()
{
    GstPad * q2SrcPad;
    q2SrcPad = gst_element_get_static_pad(q2, "src");
    gst_pad_remove_probe(q2SrcPad, probeId);
    gst_object_unref(q2SrcPad);
}

void DestroyBin()
{
    gst_element_set_state(bin, GST_STATE_NULL);
    gst_bin_remove(GST_BIN(pipeline), bin);
}

void CreateNewBin()
{

    static std::string fileLocPattern = "deneme%d.mkv";
    char buffer[12];
    memset(buffer, 0, sizeof(buffer));
    sprintf(buffer, fileLocPattern.c_str(), i++);

    //Create Muxer Element
    muxer = gst_element_factory_make("matroskamux", "MatroskaMuxer");

    //Create File Sink Element
    sink = gst_element_factory_make("filesink", buffer);
    g_object_set(G_OBJECT(sink), "location", buffer, 0);

    //Create muxsinkBin
    bin = gst_bin_new(buffer);
    g_object_set(G_OBJECT(bin), "message-forward", TRUE, 0);
    //Add a src pad to the bin
    gst_bin_add_many(GST_BIN(bin), muxer, sink, 0);

    gboolean linkState = TRUE;
    //Connect elements within muxsink_bin
    //Link: matroskamuxer -> filesink
    linkState &= gst_element_link_many(muxer, sink, 0);

    //Add this bin to pipeline
    gst_bin_add(GST_BIN(pipeline), bin);

    //Create ghostpad and manually link muxsinkBin and remaining part of the pipeline
    {
        GstPadTemplate * muxerSinkPadTemplate;


        if( !(muxerSinkPadTemplate = gst_element_class_get_pad_template(GST_ELEMENT_GET_CLASS(muxer), "video_%u")) )
        {
            std::cout << "Unable to get source pad template from muxing element" << std::endl;
        }

        //Obtain dynamic pad from element
        muxerSinkPad = gst_element_request_pad(muxer, muxerSinkPadTemplate, 0, 0);

        //Add ghostpad
        GstPad * ghostPad = gst_ghost_pad_new("sink", muxerSinkPad);
        gst_element_add_pad(bin, ghostPad);
        gst_object_unref(GST_OBJECT(muxerSinkPad));

        gst_element_sync_state_with_parent(bin);

        //Get src pad from queue element
        GstPad * queueBeforeBinSrcPad = gst_element_get_static_pad(q2, "src");

        //Link queuebeforebin to ghostpad
        if (gst_pad_link(queueBeforeBinSrcPad, ghostPad) != GST_PAD_LINK_OK )
        {

            std::cout << "QueueBeforeBin cannot be linked to MuxerSinkPad." << std::endl;
        }
        gst_object_unref(queueBeforeBinSrcPad);
    }
}

http://gstreamer-devel.966125.n4.nabble.com/Listening-on-EOS-events-for-GstBin-td4669126.html

http://gstreamer-devel.966125.n4.nabble.com/file/n4669476/main.cpp