Add or remove elements to (from) a tee dynamically

Thu Dec 24 08:33:51 PST 2015

OK, thanks
I tried this formula and I got the following error:

ERROR: from element /GstPipeline:pipeline0/GstAlsaSrc:alsasrc0: Internal
data flow error.
Additional debug info:
gstbasesrc.c(2943): gst_base_src_loop ():
/GstPipeline:pipeline0/GstAlsaSrc:alsasrc0:
streaming task paused, reason not-linked (-1)

static GstElement *pipeline;
static GstElement *src, *tee;

typedef struct
{
	GstPad *teepad;
	GstElement *queue;
	GstElement *level;
	GstElement *sink;
	gboolean removing;
} Listen;

typedef struct
{
	GstPad *teepad;
	GstElement *queue;
	GstElement *level;
	GstElement *sink;
	gboolean removing;
} Record;

static gboolean
message_cb (GstBus * bus, GstMessage * message, gpointer user_data)
{
	switch (GST_MESSAGE_TYPE (message)) {
		case GST_MESSAGE_ERROR:{
			GError *err = NULL;
			gchar *name, *debug = NULL;

			name = gst_object_get_path_string (message->src);
			gst_message_parse_error (message, &err, &debug);

			g_printerr ("ERROR: from element %s: %s\n", name, err->message);
			if (debug != NULL)
				g_printerr ("Additional debug info:\n%s\n", debug);

			g_error_free (err);
			g_free (debug);
			g_free (name);

			g_main_loop_quit (loop);
			break;
		}
		case GST_MESSAGE_WARNING:{
			GError *err = NULL;
			gchar *name, *debug = NULL;

			name = gst_object_get_path_string (message->src);
			gst_message_parse_warning (message, &err, &debug);

			g_printerr ("ERROR: from element %s: %s\n", name, err->message);
			if (debug != NULL)
				g_printerr ("Additional debug info:\n%s\n", debug);

			g_error_free (err);
			g_free (debug);
			g_free (name);
			break;
		}
		case GST_MESSAGE_EOS:
			g_print ("Got EOS\n");
			g_main_loop_quit (loop);
			break;
		default:
			break;
	}

	return TRUE;
}

static GstPadProbeReturn
record_pad_unlink_cb (GstPad * pad, GstPadProbeInfo * info, gpointer
user_data)
{
	Record *record = user_data;
	GstPad *sinkpad;

	if (!g_atomic_int_compare_and_exchange (&record->removing, FALSE, TRUE))
		return GST_PAD_PROBE_OK;

	sinkpad = gst_element_get_static_pad (record->queue, "sink");
	gst_pad_unlink (record->teepad, sinkpad);
	gst_object_unref (sinkpad);

	gst_bin_remove (GST_BIN (pipeline), record->queue);
	gst_bin_remove (GST_BIN (pipeline), record->level);
	gst_bin_remove (GST_BIN (pipeline), record->sink);

	gst_element_set_state (record->sink, GST_STATE_NULL);
	gst_element_set_state (record->level, GST_STATE_NULL);
	gst_element_set_state (record->queue, GST_STATE_NULL);

	gst_object_unref (record->queue);
	gst_object_unref (record->level);
	gst_object_unref (record->sink);

	gst_element_release_request_pad (tee, record->teepad);
	gst_object_unref (record->teepad);

	g_print ("removed\n");

	return GST_PAD_PROBE_REMOVE;
}

static void
record_pad_added_cb (GstElement * element, GstPad * pad, gpointer user_data)
{
	GstCaps *caps;
	GstStructure *s;
	const gchar *name;

	caps = gst_pad_get_current_caps (pad);
	s = gst_caps_get_structure (caps, 0);
	name = gst_structure_get_name (s);

	if (strcmp (name, "audio/x-raw") == 0) {
		GstPad *sinkpad, *teepad;
		GstElement *queue, *level, *sink;
		GstPadTemplate *templ;

		sinkpad = gst_element_get_static_pad (src, "sink");
		if (gst_pad_link (pad, sinkpad) != GST_PAD_LINK_OK) {
			g_printerr ("Failed to link src with conv\n");
			gst_object_unref (sinkpad);
			g_main_loop_quit (loop);
			return;
		}
		gst_object_unref (sinkpad);

		templ = gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS (tee),
"src_%u");
		teepad = gst_element_request_pad (tee, templ, NULL, NULL);
		queue = gst_element_factory_make ("queue", NULL);
		level = gst_element_factory_make ("level", NULL);
		sink = gst_element_factory_make ("filesink", NULL);

		gst_bin_add_many (GST_BIN (pipeline), queue, level, sink, NULL);
		gst_element_link_many (queue, level, sink, NULL);

		sinkpad = gst_element_get_static_pad (queue, "sink");
		gst_pad_link (teepad, sinkpad);
		gst_object_unref (sinkpad);
	}

	gst_caps_unref (caps);

	g_timeout_add_seconds (1, record_pad_unlink_cb, NULL);
}

static GstPadProbeReturn
listen_pad_unlink_cb (GstPad * pad, GstPadProbeInfo * info, gpointer
user_data)
{
	Listen *listen = user_data;
	GstPad *sinkpad;

	if (!g_atomic_int_compare_and_exchange (&listen->removing, FALSE, TRUE))
		return GST_PAD_PROBE_OK;

	sinkpad = gst_element_get_static_pad (listen->queue, "sink");
	gst_pad_unlink (listen->teepad, sinkpad);
	gst_object_unref (sinkpad);

	gst_bin_remove (GST_BIN (pipeline), listen->queue);
	gst_bin_remove (GST_BIN (pipeline), listen->convert);
	gst_bin_remove (GST_BIN (pipeline), listen->resample);
	gst_bin_remove (GST_BIN (pipeline), listen->sink);

	gst_element_set_state (listen->sink, GST_STATE_NULL);
	gst_element_set_state (listen->convert, GST_STATE_NULL);
	gst_element_set_state (listen->resample, GST_STATE_NULL);
	gst_element_set_state (listen->queue, GST_STATE_NULL);

	gst_object_unref (listen->queue);
	gst_object_unref (listen->convert);
	gst_object_unref (listen->resample);
	gst_object_unref (listen->sink);

	gst_element_release_request_pad (tee, listen->teepad);
	gst_object_unref (listen->teepad);

	g_print ("removed\n");

	return GST_PAD_PROBE_REMOVE;
}

static void
listen_pad_added_cb (GstElement * element, GstPad * pad, gpointer user_data)
{
	GstCaps *caps;
	GstStructure *s;
	const gchar *name;

	caps = gst_pad_get_current_caps (pad);
	s = gst_caps_get_structure (caps, 0);
	name = gst_structure_get_name (s);

	if (strcmp (name, "audio/x-raw") == 0) {
		GstPad *sinkpad, *teepad;
		GstElement *queue, *convert, *resample, *sink;
		GstPadTemplate *templ;

		sinkpad = gst_element_get_static_pad (src, "sink");
		if (gst_pad_link (pad, sinkpad) != GST_PAD_LINK_OK) {
			g_printerr ("Failed to link src with conv\n");
			gst_object_unref (sinkpad);
			g_main_loop_quit (loop);
			return;
		}
		gst_object_unref (sinkpad);

		templ = gst_element_class_get_pad_template (GST_ELEMENT_GET_CLASS (tee),
"src_%u");
		teepad = gst_element_request_pad (tee, templ, NULL, NULL);
		queue = gst_element_factory_make ("queue", NULL);
		convert = gst_element_factory_make ("audioconvert", NULL);
		resample = gst_element_factory_make ("audioresample", NULL);
		sink = gst_element_factory_make ("alsasink", NULL);

		gst_bin_add_many (GST_BIN (pipeline), queue, convert, resample, sink,
NULL);
		gst_element_link_many (queue, convert, resample, sink, NULL);

		sinkpad = gst_element_get_static_pad (queue, "sink");
		gst_pad_link (teepad, sinkpad);
		gst_object_unref (sinkpad);
	}

	gst_caps_unref (caps);
}

void create_pipeline(void)
{
	pipeline = gst_pipeline_new (NULL);
	src = gst_element_factory_make ("alsasrc", NULL);
	tee = gst_element_factory_make ("tee", NULL);

	if (!pipeline || !src || !tee) {
		g_error ("Failed to create elements");
		return;
	}

	gst_bin_add_many (GST_BIN (pipeline), src, tee, NULL);
	if (!gst_element_link_many (src, tee, NULL)) {
		g_error ("Failed to link elements");
		return;
	}

	g_signal_connect (src, "pad-added", G_CALLBACK (listen_pad_added_cb),
NULL);
	g_signal_connect (src, "pad-added", G_CALLBACK (record_pad_added_cb),
NULL);

	g_object_set (sink, "device", "hw:1", NULL);

	bus = gst_pipeline_get_bus (GST_PIPELINE (pipeline));
	gst_bus_add_signal_watch (bus);
	g_signal_connect (G_OBJECT (bus), "message", G_CALLBACK (message_cb),
NULL);
	gst_object_unref (GST_OBJECT (bus));

	gst_element_set_state (pipeline, GST_STATE_PLAYING);
}

What mistake I made in this code?

--
View this message in context: http://gstreamer-devel.966125.n4.nabble.com/Add-or-remove-elements-to-from-a-tee-dynamically-tp4675008p4675019.html
Sent from the GStreamer-devel mailing list archive at Nabble.com.