In gstreamer cant remove tee section after EOS

Question

I am trying to create a webcam on an embedded device and learn gstreamer c implementation at the same time. i have dealt with gstreamer launch pipelines for a while so i am somewhat familiar already with gstreamer.

my end goal is to eventually have a pipeline that will dynamically stream video, record video and save pictures all from external commands. I've started small with my implementation and right now I'm focusing on being able to take a picture in one branch of a tee while the other branch is still flowing data. the other branch is just a fakesink right now but eventually it will be an h264 encoder with mux and audio saving videos.

here is a simple view of my pipeline: v4l2src ! capsfilter ! tee ! queue ! fakesink tee. ! queue ! videoconvert ! pngenc ! filesink

my idea was to dynamically add the picture portion of the pipeline while its running. the flow of my program goes like this:
picture event is triggered (currently a simple timer)-> add blocking probe on tee -> add picture pipeline and link it to tee -> set to playing -> set blocking probe on filesink to verify it has received data -> send EOS down the pipeline starting at the videoconvert -> set blocking probe on tee pad linked to picture pipeline -> set the picture pipeline to null and remove it and the tee pad

when the program executes, the eos probe on the tee pad for the picture pipeline is never called and instead the whole pipeline goes to EOS and i get an internal data stream error and no picture.

i want to make sure the filesink only gets 1 buffer as i cant stop the v4l2src stream or give it a num-buffers=1. i guess my problem right now is: how do i verify the filesink gets only one buffer? which pad should i send the EOS event on in order for it to properly save the picture? and lastly, how do i make sure only this one branch sees the EOS?

ive poured over all of the gstreamer tutorials and SO questions but most are either not answered or havent helped my situation.

here is my code:

#include <QDebug>
#include <QTimer>
#include "gstpipeline.hpp"
#include "gsttypes.hpp"

using namespace INSP_GST_TYPES;

gstpipeline::gstpipeline()
: mV4l2Src(NULL)
, mEncoder(NULL)
, mPngEncoder(NULL)
, mVideoFileSink(NULL)
, mPictureFileSink(NULL)
, mRawCapsFilter(NULL)
, mEncodedCapsFilter(NULL)
, mEncoderVideoConvert(NULL)
, mPngVideoConvert(NULL)
, mEncoderQueue(NULL)
, mMatroskaMux(NULL)
, mPipeline(NULL)
{
}

void gstpipeline::init()
{
    mV4l2Src = gst_element_factory_make("v4l2src", V4L2SOURCE_NAME);
    mRawCapsFilter = gst_element_factory_make("capsfilter", RAW_CAPS_NAME);
    mRawFakesinkQueue = gst_element_factory_make("queue", RAW_FAKESINK_QUEUE_NAME);
    mRawFakeSink = gst_element_factory_make("fakesink", RAW_FAKESINK_NAME);
    mRawTee = gst_element_factory_make("tee", RAW_TEE_NAME);
    mPipeline = gst_pipeline_new(PIPELINE_NAME);

    mRawCaps = gst_caps_new_simple("video/x-raw",
                                   "format", G_TYPE_STRING, "NV12",
                                   "width", G_TYPE_INT, 1280,
                                   "height", G_TYPE_INT, 720,
                                   "framerate", GST_TYPE_FRACTION, 30, 1,
                                   NULL);

    g_object_set(mRawCapsFilter, "caps", mRawCaps, NULL);

    if(!mPipeline || !mV4l2Src || !mRawCapsFilter || !mRawTee || !mRawFakesinkQueue || !mRawFakeSink)
    {
        qCritical() << "Failed to create main gst elements";
        return;
    }
    else
    {
        qWarning() << "created the initial pipeline";
    }

    linkRawPipeline();
}

void gstpipeline::linkRawPipeline()
{
    gst_bin_add_many(GST_BIN(mPipeline), mV4l2Src, mRawCapsFilter, mRawTee, mRawFakesinkQueue, mRawFakeSink, NULL);
    g_object_set(mPipeline, "message-forward", TRUE, NULL);
    if(gst_element_link_many(mV4l2Src, mRawCapsFilter, mRawTee, NULL) != TRUE)
    {
        qCritical() << "Failed to link raw pipeline";
        return;
    }
    if(gst_element_link_many(mRawFakesinkQueue, mRawFakeSink, NULL) != TRUE)
    {
        qCritical() << "Failed to link fakesink pipeline";
        return;
    }

    /* Manually link the Tee, which has "Request" pads */
    GstPad* tee_fakesink_pad = gst_element_get_request_pad (mRawTee, "src_%u");
    qWarning ("Obtained request pad %s for fakesink branch.", gst_pad_get_name (tee_fakesink_pad));
    GstPad* raw_queue_pad = gst_element_get_static_pad (mRawFakesinkQueue, "sink");
    if (gst_pad_link (tee_fakesink_pad, raw_queue_pad) != GST_PAD_LINK_OK)
    {
      qCritical ("raw Tee could not be linked.");
    }
    gst_object_unref(tee_fakesink_pad);
    gst_object_unref(raw_queue_pad);

    if (gst_element_set_state (mPipeline, GST_STATE_PLAYING) == GST_STATE_CHANGE_FAILURE)
    {
        qCritical() << "Unable to set the pipeline to the ready state";
        gst_object_unref (mPipeline);
    }
    else
    {
        qWarning() << "set pipeline to playing";

        GMainLoop* loop = g_main_loop_new (NULL, FALSE);
        gst_bus_add_watch (GST_ELEMENT_BUS (mPipeline), sMainBusCallback, loop);

        QTimer::singleShot(1000, this, SLOT(onBusTimeoutExpired()));
    }
}

void gstpipeline::onBusTimeoutExpired()
{
    blockRawPipeline();
}

void gstpipeline::blockRawPipeline()
{
    qWarning() << "Blocking raw pipeline";
    GstPad* srcpad = gst_element_get_static_pad(mRawFakesinkQueue, SRC_PAD);

    gst_pad_add_probe(srcpad,
                (GstPadProbeType)(GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM | GST_PAD_PROBE_TYPE_IDLE),
                sRawFakesinkQueueBlockedCallback, NULL, NULL);

    g_object_unref(srcpad);
    qWarning() << "added fakesink queue probe";
}

GstPadProbeReturn gstpipeline::sRawFakesinkQueueBlockedCallback(GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
{
    gst_pad_remove_probe (pad, GST_PAD_PROBE_INFO_ID (info));

    //create the picturesink pipeline and link it to a new src pad on the raw tee
    mPictureQueue = gst_element_factory_make("queue", RAW_PICTURE_QUEUE_NAME);
    mPngEncoder = gst_element_factory_make("pngenc", PNG_ENC_NAME);
    mPictureFileSink = gst_element_factory_make("filesink", PICTURESINK_NAME);
    mPngVideoConvert =  gst_element_factory_make("videoconvert", VIDEOCONVERT_PNG_NAME);

    if(!mPngEncoder || !mPictureFileSink || !mPngVideoConvert)
    {
        qCritical() << "failed to make picturesink elements";
    }

    g_object_set(G_OBJECT (mPictureFileSink), "location", "/mnt/userdata/pipelinetest.png", NULL);

    gst_bin_add_many (GST_BIN (mPipeline), mPictureQueue, mPngVideoConvert,
                      mPngEncoder, mPictureFileSink, NULL);

    if(gst_element_link_many(mPictureQueue, mPngVideoConvert, mPngEncoder, mPictureFileSink, NULL) != TRUE)
    {
        qCritical() << "failed to link picture pipeline";
    }

    GstPad* tee_picturesink_pad = gst_element_get_request_pad (mRawTee, "src_%u");
    qWarning ("Obtained request pad %s for picturesink branch.", gst_pad_get_name (tee_picturesink_pad));
    GstPad* raw_picture_queue_pad = gst_element_get_static_pad (mPictureQueue, "sink");
    if (gst_pad_link (tee_picturesink_pad, raw_picture_queue_pad) != GST_PAD_LINK_OK)
    {
      qCritical ("picture Tee could not be linked.");
    }

    gst_element_sync_state_with_parent(mPictureQueue);
    gst_element_sync_state_with_parent(mPngVideoConvert);
    gst_element_sync_state_with_parent(mPngEncoder);
    gst_element_sync_state_with_parent(mPictureFileSink);

    qWarning() << "done adding picturesink";

    //set data block to see when the filesink gets data so we can send an EOS
    GstPad* srcpad = gst_element_get_static_pad(mPictureFileSink, SINK_PAD);

    gst_pad_add_probe(srcpad,  (GstPadProbeType)(GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM),
                       sPictureSinkDownstreamBlockProbe, NULL, NULL);

    g_object_unref(srcpad);

    return GST_PAD_PROBE_DROP;
}

GstPadProbeReturn gstpipeline::sPictureSinkDownstreamBlockProbe(GstPad *pad, GstPadProbeInfo *info, gpointer user_data)
{
    gst_pad_remove_probe (pad, GST_PAD_PROBE_INFO_ID (info));

    //this is a data blocking pad probe on picture filesink
    qWarning() << "setting the EOS event probe on the picturesink";
    GstPad* srcpad = gst_element_get_static_pad(mPictureQueue, SRC_PAD);
    gst_pad_add_probe(pad, (GstPadProbeType)(GST_PAD_PROBE_TYPE_BLOCK | GST_PAD_PROBE_TYPE_EVENT_DOWNSTREAM),sPictureSinkEOSCallback, NULL, NULL);
    g_object_unref(srcpad);

    qWarning() << "sending eos through videoconvert";
    gst_element_send_event(mPngVideoConvert, gst_event_new_eos());

    qWarning() << "exiting pad probe";
    return GST_PAD_PROBE_PASS;
}

GstPadProbeReturn gstpipeline::sPictureSinkEOSCallback(GstPad *pad, GstPadProbeInfo *info, gpointer user_data)
{
    gst_pad_remove_probe (pad, GST_PAD_PROBE_INFO_ID (info));
    if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_DATA (info)) == GST_EVENT_EOS)
    {
        qWarning() << "setting raw queue pad block";
        GstPad* srcpad = gst_element_get_static_pad(mPictureQueue, SRC_PAD);
        gst_pad_add_probe(pad, (GstPadProbeType)(GST_PAD_PROBE_TYPE_IDLE),sRawQueueBlockedCallback, NULL, NULL);
        g_object_unref(srcpad);
    }
    else
    {
        qCritical() << "picturesink pad probe is NOT EOS";
    }

    return GST_PAD_PROBE_HANDLED;
}

GstPadProbeReturn gstpipeline::sRawQueueBlockedCallback(GstPad *pad, GstPadProbeInfo *info, gpointer user_data)
{
    if (GST_EVENT_TYPE (GST_PAD_PROBE_INFO_DATA (info)) == GST_EVENT_EOS)
    {
        gst_pad_remove_probe (pad, GST_PAD_PROBE_INFO_ID (info));

        gst_element_set_state(mPictureFileSink, GST_STATE_NULL);
        gst_element_set_state(mPngEncoder, GST_STATE_NULL);
        gst_element_set_state(mPngVideoConvert, GST_STATE_NULL);
        gst_element_set_state(mPictureQueue, GST_STATE_NULL);

        //unlink the picture pipeline from the src pad of the raw tee and remove that pad
        GstPad* tee_picturesink_pad = gst_element_get_static_pad(mRawTee, "src_1");
        qWarning ("Obtained request pad %s for picturesink branch.", gst_pad_get_name (tee_picturesink_pad));
        GstPad* raw_picture_queue_pad = gst_element_get_static_pad (mPictureQueue, "sink");
        if (gst_pad_unlink (tee_picturesink_pad, raw_picture_queue_pad) != GST_PAD_LINK_OK)
        {
          qCritical ("picture Tee could not be linked.");
        }
        if(gst_element_remove_pad(mRawTee, tee_picturesink_pad) != TRUE)
        {
            qCritical("could not remove raw tee pad");
        }
        g_object_unref(tee_picturesink_pad);
        g_object_unref(raw_picture_queue_pad);

        gst_bin_remove_many(GST_BIN(mPipeline), mPictureQueue, mPngVideoConvert, mPngEncoder, mPictureFileSink, NULL);

        qWarning() << "we have set the fakesink back up";
    }
    else
    {
        qCritical() << "picturesink pad probe is NOT EOS";
    }
    return GST_PAD_PROBE_PASS;
}

gboolean gstpipeline::sMainBusCallback (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_clear_error (&err);
      g_free (dbg);
      g_main_loop_quit (loop);
  }
      break;
    case GST_MESSAGE_EOS:
      g_print ("we reached EOS\n");
      g_main_loop_quit (loop);
      break;
    default:
//      g_print ("msg: %s\n", GST_MESSAGE_TYPE_NAME(msg));
      break;
  }
}

Answer 1

so i managed to figure this out myself. here are the steps i took to get this working:
1. blocking probe on the fakesink queue
2. add the picture pipeline
3. put a blocking data probe on the picture files sink
4. wait until a segment buffer reaches the filesink
5. put a blocking probe on the picture piplines queue
6. in the queue blocking probe, send eos event and remove the picture pipeline