sinker.c

//sinker.c -- test program to add and remove sinks to/from a pipeline.

#include <stdlib.h>
#include <gst/gst.h>

#define DISPLAY ":0"

// Returns the short name for an enum's value, or NULL if not found.
static gchar const * enum_nick
  ( GType enum_id
  , gint  value
  )
  { GEnumClass  * gec = G_ENUM_CLASS(g_type_class_ref(enum_id));
    GEnumValue  * val = gec->values;
    gchar const * ret = NULL;
    gint          i;

    for(i=0; i < gec->n_values; i++)
      if( val[i].value == value )
	{
	  ret = val[i].value_nick;
	  break;
	}
    g_type_class_unref(gec);
    return ret;
  }

#define STREAM_STATUS_TYPE_NAME(ss) (enum_nick(GST_TYPE_STREAM_STATUS_TYPE,ss))

static gboolean bus_callback
  ( GstBus	* bus
  , GstMessage  * msg
  , gpointer      data
  )
  { GMainLoop   * loop = data;
    gboolean      quit = FALSE;	// do we need to quit the program?
    gchar const * src  = GST_MESSAGE_SRC_NAME(msg);
    gchar       * bug  = NULL;
    GError      * err  = NULL;

    switch( GST_MESSAGE_TYPE(msg) )
      {
	case GST_MESSAGE_EOS:
	  GST_LOG("End of stream");
	  quit = TRUE;
	  break;

#if 0
	case GST_MESSAGE_STATE_CHANGED:
	  { GstState old;
	    GstState new;
	    GstState pending;

	    gst_message_parse_state_changed(msg,&old,&new,&pending);

	    if( pending != GST_STATE_VOID_PENDING )
	      g_print
		( "State Change: %-8s: %-8s -> %-8s -> %s\n"
		, GST_MESSAGE_SRC_NAME(msg)
		, gst_element_state_get_name(old)
		, gst_element_state_get_name(new)
		, gst_element_state_get_name(pending)
		);
	    else
	      g_print
		( "State Change: %-8s: %-8s -> %-8s\n"
		, GST_MESSAGE_SRC_NAME(msg)
		, gst_element_state_get_name(old)
		, gst_element_state_get_name(new)
		);
	    break;
	  }

	case GST_MESSAGE_STREAM_STATUS:
	  { GstStreamStatusType   sstype;
	    GstElement          * owner;

	    gst_message_parse_stream_status(msg,&sstype,&owner);
	    g_print
	      ( "Stream Status: %s : %s\n"
	      , GST_ELEMENT_NAME(owner)
	      , STREAM_STATUS_TYPE_NAME(sstype)
	      );
	    break;
	  }
#endif

	case GST_MESSAGE_CLOCK_LOST:
	  g_print("Clock Lost!\n");
	  break;

	case GST_MESSAGE_NEW_CLOCK:
	  g_printf("New Clock Chosen!\n");
	  break;

	case GST_MESSAGE_ERROR:
	  gst_message_parse_error(msg,&err,&bug);
	  GST_ERROR
	    ( "Error %d.%d: %s (%s)\n"
	    , err->domain,  err->code
	    , err->message, bug
	    );
	  g_error_free(err);
	  g_free(msg);
	  quit = TRUE;
	  break;

	case GST_MESSAGE_WARNING:
	  gst_message_parse_warning(msg,&err,&bug);
	  GST_WARNING("Warning: %s: %s (%s)",src,err->message,bug);
	  g_error_free(err);
	  g_free(bug);
	  break;

	case GST_MESSAGE_INFO:
	  gst_message_parse_info(msg,&err,&bug);
	  GST_INFO("Info: %s: %s (%s)",src, err->message,bug);
	  g_print("Info: %s: %s\n",src, err->message);
	  g_error_free(err);
	  g_free(bug);
	  break;

	default:
	  break;
      }
    if( quit )
      g_main_loop_quit(loop);
    return TRUE;
  }

typedef struct CBData
  { GMainLoop   * loop;		// Loop context
    GstBin      * pipeline;	// Our pipeline
    GstElement  * tee;		// Tee junction in pipeline
    GstElement  * out;		// Output pipeline
    GstElement  * sink;		// Sink at end of output pipeline.
    guint         count;	// Number of add/remove cycles to do
    GMutex        mutex;
    GCond         cond;
  }
CBData;

static GstPadProbeReturn pad_wait_and_destruct_callback
  ( GstPad          * pad
  , GstPadProbeInfo * info
  , gpointer          data
  )
  { CBData          * cb      = data;

    g_print("Destruct wakes..\n");

    if(GST_EVENT_TYPE(GST_PAD_PROBE_INFO_DATA(info)) != GST_EVENT_FLUSH_START)
      return GST_PAD_PROBE_OK;

    g_print("Destruction Probe Activated.\n");
    gst_pad_remove_probe(pad,GST_PAD_PROBE_INFO_ID(info));

    gst_element_set_state(cb->out,GST_STATE_NULL);

    // NOTE: gst_bin_remove unparents the stream, which counts as an
    // unref. This is sufficient to free the stream.
    gst_bin_remove(cb->pipeline,cb->out);

    // announce we did it.
    g_print("Signaling sink removal\n");
    cb->sink = NULL;
    cb->out  = NULL;
    g_cond_signal(&cb->cond);
    return GST_PAD_PROBE_OK;
  }

static GstPadProbeReturn pad_block_and_sever_callback
  ( GstPad          * pad	// source pad of tee element
  , GstPadProbeInfo * info
  , gpointer          data
  )
  { CBData          * cb      = data;
    GstPad	    * sinkpad = gst_pad_get_peer(pad);
    GstElement      * parent  = gst_pad_get_parent_element(pad);

    g_print("Block probe activated\n");

    // We're a one-shot, so remove us from this pad.
    gst_pad_remove_probe(pad, GST_PAD_PROBE_INFO_ID(info));

    // Unlink the sink from our pad.
    gst_pad_unlink(pad,sinkpad);

    // Ask our parent (TEE element) to remove this request pad
    gst_element_release_request_pad(parent,pad);

    // Note: This is slightly futile in that we are setting a probe
    // on a sink pad and then immediately triggering it, because in
    // this case 'sinksink' and 'sinkpad' are the same pads, but in the
    // case of a longer output pipeline than a single element, this
    // becomes a useful operation.

    GstPad * sinksink = gst_element_get_static_pad(cb->sink,"sink");

    g_print("Adding destruction probe\n");
    gst_pad_add_probe
      ( sinksink
      , GST_PAD_PROBE_TYPE_EVENT_FLUSH
      , pad_wait_and_destruct_callback
      , cb
      , NULL
      );

    g_print("Activating destruction probe\n");
    gst_pad_send_event(sinkpad,gst_event_new_flush_start());

    // And now we don't need these various references any more...
    gst_object_unref(sinksink);
    gst_object_unref(sinkpad);
    gst_object_unref(parent);
    return GST_PAD_PROBE_OK;
  }

static void pipeline_remove_output
  ( CBData * cb
  )
  {
    g_mutex_lock( &cb->mutex );

    GstPad * sinkpad = gst_element_get_static_pad(cb->out,"sink");
    GstPad * srcpad  = gst_pad_get_peer(sinkpad);

    g_print("Adding Block Probe.\n");
    gst_pad_add_probe
      ( srcpad
      , GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM
      , pad_block_and_sever_callback
      , cb
      , NULL
      );

    g_print("Waiting for removal done signal\n");
    while( cb->out )
      g_cond_wait(&cb->cond,&cb->mutex);
    g_print("Received removal done signal\n");

    g_mutex_unlock( &cb->mutex );
  }

static void pipeline_add_output
  ( CBData * cb
  )
  {
    g_mutex_lock( &cb->mutex );
    GstElement * bin   = gst_bin_new("outbin");
    GstElement * queue = gst_element_factory_make("queue",       "outqueue");
    GstElement * sink  = gst_element_factory_make("xvimagesink", "xsink"   );

    g_object_set(sink,"display",DISPLAY,NULL);

    gst_bin_add_many(GST_BIN(bin),queue,sink,NULL);
    gst_element_link(queue,sink);

    GstPad *qsink = gst_element_get_static_pad(queue,"sink");
    GstPad *gsink = gst_ghost_pad_new("sink",qsink);

    gst_pad_set_active(gsink,TRUE);
    gst_element_add_pad(bin,gsink);
    gst_object_unref(qsink);

    gst_bin_add(cb->pipeline,bin);
    if( !gst_element_link(cb->tee,bin) )
      { g_print("Failed to link new bin.\n"); exit(-1); }

    gst_element_sync_state_with_parent(bin);

    cb->out  = bin;
    cb->sink = sink;

    g_mutex_unlock(&cb->mutex);
  }

static gboolean timed_callback
  ( gpointer   data
  )
  { CBData   * cb = data;

    if( cb->count == 0 )
      {
	g_main_loop_quit(cb->loop);
	return FALSE;
      }
    cb->count--;
    if( cb->out )
      {
	g_print("Removing Output.\n");
	pipeline_remove_output(cb);
	g_print("Output Removed.\n");
      }
    else
      {
	g_print("Adding Output.\n");
	pipeline_add_output(cb);
	g_print("Output Added.\n");
      }
    return TRUE;
  };


gint main
  ( gint    argc
  , gchar * argv[]
  )
  {
    /* init GStreamer */
    gst_init(&argc, &argv);

    GMainLoop  * loop = g_main_loop_new(NULL, FALSE);

    /* create a new pipeline to hold the elements */
    GstElement * pipeline = gst_pipeline_new("pipe");
    GstBus     * bus      = gst_pipeline_get_bus( GST_PIPELINE(pipeline) );

    gst_bus_add_watch(bus, bus_callback, loop);
    gst_object_unref(bus);

    /* create basic pipeline elments */
    GstElement * vidsrc   = gst_element_factory_make("videotestsrc", "source");
    GstElement * tee      = gst_element_factory_make("tee",          "tee"   );
    GstElement * queue    = gst_element_factory_make("queue",        "fqueue");
    GstElement * fakesink = gst_element_factory_make("fakesink",     "fake"  );

    /* initial setup */
    gst_bin_add_many(GST_BIN(pipeline), vidsrc, tee, queue, fakesink, NULL);
    gst_element_link_many(vidsrc, tee, fakesink, NULL);

    GstStateChangeReturn sts = gst_element_set_state
      ( pipeline
      , GST_STATE_PLAYING
      );

    if( sts == GST_STATE_CHANGE_FAILURE )
      {
	GST_ERROR("Pipeline could not start playing");
	exit(-1);
      }
    else if( sts == GST_STATE_CHANGE_ASYNC )
      {
	g_print("Waiting for pipeline to change state.\n");
	sts = gst_element_get_state(pipeline, NULL, NULL, 5*GST_SECOND);
	if( sts == GST_STATE_CHANGE_FAILURE )
	  {
	    GST_ERROR("Pipeline Asynchronously failed");
	    g_print("Pipeline failure!\n");
	    exit(-1);
	  }
      }
    g_print("Pipeline in Run State.\n");

    CBData cb = {loop, GST_BIN(pipeline), tee, NULL, NULL, 1, FALSE, 0, 0};

    g_mutex_init(&cb.mutex);
    g_cond_init(&cb.cond);

    g_timeout_add_seconds(2, timed_callback, &cb);

    g_print("Starting Run Loop\n");
    g_main_loop_run(loop);
    g_print("Exiting Run Loop\n");

    /* unset */
    gst_element_set_state(GST_ELEMENT(pipeline), GST_STATE_NULL);
    gst_object_unref(GST_OBJECT(pipeline));
    g_cond_clear(&cb.cond);
    g_mutex_clear(&cb.mutex);
    return 0;
  }