tests/examples/streamiddemux/streamiddemux-stream.c

#include <gst/gst.h>

#define NUM_STREAM 13

typedef struct _App App;

struct _App
{
  GstElement *pipeline;
  GstElement *audiotestsrc[NUM_STREAM];
  GstElement *audioconvert[NUM_STREAM];
  GstElement *capsfilter[NUM_STREAM];
  GstElement *vorbisenc[NUM_STREAM];
  GstElement *oggmux[NUM_STREAM];
  GstElement *funnel;
  GstElement *demux;
  GstElement *stream_synchronizer;
  GstElement *queue[NUM_STREAM];
  GstElement *filesink[NUM_STREAM];

  gboolean pad_blocked[NUM_STREAM];
  GstPad *queue_srcpad[NUM_STREAM];
  gulong blocked_id[NUM_STREAM];
};

App s_app;

gint pad_added_cnt = 0;

static gboolean
bus_call (GstBus * bus, GstMessage * msg, gpointer data)
{
  GMainLoop *loop = (GMainLoop *) data;

  switch (GST_MESSAGE_TYPE (msg)) {
    case GST_MESSAGE_EOS:{
      g_main_loop_quit (loop);
      break;
    }
    case GST_MESSAGE_ERROR:{
      g_main_loop_quit (loop);
      break;
    }
    default:
      break;
  }
  return TRUE;
}

static void
set_blocked (App * app, gboolean blocked)
{
  gint i = 0;

  for (i = 0; i < NUM_STREAM; i++) {
    gst_pad_remove_probe (app->queue_srcpad[i], app->blocked_id[i]);
  }
}

static void
sink_do_reconfigure (App * app)
{
  gint i = 0;
  GstPad *filesink_sinkpad[NUM_STREAM];
  GstPad *sync_sinkpad[NUM_STREAM];
  GstPad *sync_srcpad[NUM_STREAM];
  GstIterator *it;
  GValue item = G_VALUE_INIT;

  for (i = 0; i < NUM_STREAM; i++) {
    sync_sinkpad[i] =
        gst_element_get_request_pad (app->stream_synchronizer, "sink_%u");
    it = gst_pad_iterate_internal_links (sync_sinkpad[i]);
    g_assert (it);
    gst_iterator_next (it, &item);
    sync_srcpad[i] = g_value_dup_object (&item);
    g_value_unset (&item);

    filesink_sinkpad[i] = gst_element_get_static_pad (app->filesink[i], "sink");

    gst_pad_link_full (app->queue_srcpad[i], sync_sinkpad[i],
        GST_PAD_LINK_CHECK_NOTHING);
    gst_pad_link_full (sync_srcpad[i], filesink_sinkpad[i],
        GST_PAD_LINK_CHECK_NOTHING);
  }
  gst_iterator_free (it);

}

static GstPadProbeReturn
blocked_cb (GstPad * blockedpad, GstPadProbeInfo * info, gpointer user_data)
{
  App *app = user_data;
  gint i = 0;
  gboolean all_pads_blocked = TRUE;

  for (i = 0; i < NUM_STREAM; i++) {
    if (blockedpad == app->queue_srcpad[i])
      app->pad_blocked[i] = TRUE;
  }

  for (i = 0; i < NUM_STREAM; i++) {
    if (app->queue_srcpad[i] == FALSE) {
      all_pads_blocked = FALSE;
      break;
    }
  }

  if (all_pads_blocked == TRUE) {
    sink_do_reconfigure (app);
    set_blocked (app, FALSE);
  }

  return GST_PAD_PROBE_OK;
}

static void
src_pad_added_cb (GstElement * demux, GstPad * pad, App * app)
{
  GstPad *queue_sinkpad[NUM_STREAM];

  queue_sinkpad[pad_added_cnt] =
      gst_element_get_static_pad (app->queue[pad_added_cnt], "sink");
  gst_pad_link_full (pad, queue_sinkpad[pad_added_cnt],
      GST_PAD_LINK_CHECK_NOTHING);

  app->queue_srcpad[pad_added_cnt] =
      gst_element_get_static_pad (app->queue[pad_added_cnt], "src");
  app->blocked_id[pad_added_cnt] =
      gst_pad_add_probe (app->queue_srcpad[pad_added_cnt],
      GST_PAD_PROBE_TYPE_BLOCK_DOWNSTREAM, blocked_cb, app, NULL);

  pad_added_cnt++;
}

gint
main (gint argc, gchar * argv[])
{
  App *app = &s_app;

  GMainLoop *loop = NULL;
  GstBus *bus;
  guint bus_watch_id;

  GstPad *funnel_sinkpad[NUM_STREAM];
  GstPad *funnel_srcpad;
  GstPad *demux_sinkpad;
  GstPad *oggmux_srcpad[NUM_STREAM];

  guint stream_cnt = 0;
  GstCaps *caps;

  gst_init (&argc, &argv);

  app->pipeline = gst_pipeline_new ("pipeline");

  for (stream_cnt = 0; stream_cnt < NUM_STREAM; stream_cnt++) {
    app->audiotestsrc[stream_cnt] =
        gst_element_factory_make ("audiotestsrc", NULL);
    app->audioconvert[stream_cnt] =
        gst_element_factory_make ("audioconvert", NULL);
    app->capsfilter[stream_cnt] = gst_element_factory_make ("capsfilter", NULL);
    app->vorbisenc[stream_cnt] = gst_element_factory_make ("vorbisenc", NULL);
    app->oggmux[stream_cnt] = gst_element_factory_make ("oggmux", NULL);
  }

  app->funnel = gst_element_factory_make ("funnel", NULL);
  app->demux = gst_element_factory_make ("streamiddemux", NULL);
  app->stream_synchronizer =
      gst_element_factory_make ("streamsynchronizer", NULL);

  caps = gst_caps_from_string ("audio/x-raw,channels=1;");

  stream_cnt = 0;

  for (stream_cnt = 0; stream_cnt < NUM_STREAM; stream_cnt++) {
    app->queue[stream_cnt] = gst_element_factory_make ("queue", NULL);
    app->filesink[stream_cnt] = gst_element_factory_make ("filesink", NULL);

    g_object_set (app->audiotestsrc[stream_cnt], "wave", stream_cnt,
        "num-buffers", 2000, NULL);
    g_object_set (app->capsfilter[stream_cnt], "caps", caps, NULL);
    g_object_set (app->filesink[stream_cnt], "location",
        g_strdup_printf ("filesink_%d.ogg", stream_cnt), NULL);
  }

  stream_cnt = 0;

  g_signal_connect (app->demux, "pad-added", G_CALLBACK (src_pad_added_cb),
      app);

  loop = g_main_loop_new (NULL, FALSE);

  bus = gst_element_get_bus (app->pipeline);
  bus_watch_id = gst_bus_add_watch (bus, bus_call, loop);
  g_object_unref (bus);

  for (stream_cnt = 0; stream_cnt < NUM_STREAM; stream_cnt++) {
    gst_bin_add_many (GST_BIN (app->pipeline), app->audiotestsrc[stream_cnt],
        app->audioconvert[stream_cnt], app->capsfilter[stream_cnt],
        app->vorbisenc[stream_cnt], app->oggmux[stream_cnt],
        app->queue[stream_cnt], app->filesink[stream_cnt], NULL);
    if (stream_cnt == 0) {
      gst_bin_add_many (GST_BIN (app->pipeline), app->funnel, app->demux,
          app->stream_synchronizer, NULL);
    }
  }

  stream_cnt = 0;

  for (stream_cnt = 0; stream_cnt < NUM_STREAM; stream_cnt++) {
    gst_element_link_many (app->audiotestsrc[stream_cnt],
        app->audioconvert[stream_cnt], app->capsfilter[stream_cnt],
        app->vorbisenc[stream_cnt], app->oggmux[stream_cnt], NULL);
  }

  stream_cnt = 0;

  for (stream_cnt = 0; stream_cnt < NUM_STREAM; stream_cnt++) {
    funnel_sinkpad[stream_cnt] =
        gst_element_get_request_pad (app->funnel, "sink_%u");
    oggmux_srcpad[stream_cnt] =
        gst_element_get_static_pad (app->oggmux[stream_cnt], "src");
    gst_pad_link (oggmux_srcpad[stream_cnt], funnel_sinkpad[stream_cnt]);
  }

  funnel_srcpad = gst_element_get_static_pad (app->funnel, "src");

  demux_sinkpad = gst_element_get_static_pad (app->demux, "sink");
  gst_pad_link (funnel_srcpad, demux_sinkpad);

  gst_element_set_state (app->pipeline, GST_STATE_PLAYING);
  g_main_loop_run (loop);

  gst_element_set_state (app->pipeline, GST_STATE_NULL);
  g_object_unref (app->pipeline);
  g_source_remove (bus_watch_id);
  g_main_loop_unref (loop);

  return 0;
}