diff options
| -rw-r--r-- | ChangeLog | 20 | ||||
| -rw-r--r-- | examples/app/.gitignore | 1 | ||||
| -rw-r--r-- | examples/app/Makefile.am | 9 | ||||
| -rw-r--r-- | examples/app/appsink-src.c | 190 | ||||
| -rw-r--r-- | gst-libs/gst/app/gstappsrc.c | 76 | ||||
| -rw-r--r-- | gst-libs/gst/app/gstappsrc.h | 1 |
6 files changed, 280 insertions, 17 deletions
@@ -1,3 +1,23 @@ +2008-06-12 Wim Taymans <wim.taymans@collabora.co.uk> + + * examples/app/.cvsignore: + * examples/app/Makefile.am: + * examples/app/appsink-src.c: (on_new_buffer_from_source), + (on_source_message), (on_sink_message), (main): + Add beefed up example app from bug #413418. It now also uses appsink + instead of fakesink for more ultimate coolness. + + * gst-libs/gst/app/gstappsrc.c: (gst_app_src_class_init), + (gst_app_src_init), (gst_app_src_set_property), + (gst_app_src_get_property), (gst_app_src_unlock), + (gst_app_src_unlock_stop), (gst_app_src_create), + (gst_app_src_set_max_bytes), (gst_app_src_push_buffer), + (gst_app_src_end_of_stream): + * gst-libs/gst/app/gstappsrc.h: + Add block property to allow push based implementation to block when we + fill up the appsrc queues. + Emit the enough-data signal while releasing our lock. + 2008-06-12 Stefan Kost <ensonic@users.sf.net> * examples/app/.cvsignore: diff --git a/examples/app/.gitignore b/examples/app/.gitignore index eeae0e65..8a2c7615 100644 --- a/examples/app/.gitignore +++ b/examples/app/.gitignore @@ -3,3 +3,4 @@ appsrc-ra appsrc-seekable appsrc-stream appsrc-stream2 +appsink-src diff --git a/examples/app/Makefile.am b/examples/app/Makefile.am index fc4b9419..4f3df777 100644 --- a/examples/app/Makefile.am +++ b/examples/app/Makefile.am @@ -1,6 +1,6 @@ noinst_PROGRAMS = appsrc_ex appsrc-stream appsrc-stream2 appsrc-ra \ - appsrc-seekable + appsrc-seekable appsink-src appsrc_ex_SOURCES = appsrc_ex.c appsrc_ex_CFLAGS = $(GST_CFLAGS) $(GCONF_CFLAGS) @@ -23,3 +23,10 @@ appsrc_ra_LDFLAGS = $(GST_LIBS) appsrc_seekable_SOURCES = appsrc-seekable.c appsrc_seekable_CFLAGS = $(GST_CFLAGS) $(GCONF_CFLAGS) appsrc_seekable_LDFLAGS = $(GST_LIBS) + +appsink_src_SOURCES = appsink-src.c +appsink_src_CFLAGS = $(GST_CFLAGS) $(GCONF_CFLAGS) +appsink_src_LDFLAGS = \ + $(GST_LIBS) \ + $(top_builddir)/gst-libs/gst/app/libgstapp-@GST_MAJORMINOR@.la + diff --git a/examples/app/appsink-src.c b/examples/app/appsink-src.c new file mode 100644 index 00000000..6472af30 --- /dev/null +++ b/examples/app/appsink-src.c @@ -0,0 +1,190 @@ +#include <gst/gst.h> + +#include <string.h> + +#include <gst/app/gstappsrc.h> +#include <gst/app/gstappsink.h> +#include <gst/app/gstappbuffer.h> + +/* these are the caps we are going to pass through the appsink and appsrc */ +const gchar *audio_caps = + "audio/x-raw-int,channels=1,rate=8000,signed=(boolean)true,width=16,depth=16,endianness=1234"; + +typedef struct +{ + GMainLoop *loop; + GstElement *source; + GstElement *sink; +} ProgramData; + +/* called when the appsink notifies us that there is a new buffer ready for + * processing */ +static void +on_new_buffer_from_source (GstElement * elt, ProgramData * data) +{ + guint size; + gpointer raw_buffer; + GstBuffer *app_buffer, *buffer; + GstElement *source; + + /* get the buffer from appsink */ + buffer = gst_app_sink_pull_buffer (GST_APP_SINK (elt)); + + /* turn it into an app buffer, it's not really needed, we could simply push + * the retrieved buffer from appsink into appsrc just fine. */ + size = GST_BUFFER_SIZE (buffer); + g_print ("Pushing a buffer of size %d\n", size); + raw_buffer = g_malloc0 (size); + memcpy (raw_buffer, GST_BUFFER_DATA (buffer), size); + app_buffer = gst_app_buffer_new (raw_buffer, size, g_free, raw_buffer); + + /* we don't need the appsink buffer anymore */ + gst_buffer_unref (buffer); + + /* newer basesrc will set caps for use automatically but it does not really + * hurt to set it on the buffer again */ + gst_buffer_set_caps (app_buffer, GST_BUFFER_CAPS (buffer)); + source = gst_bin_get_by_name (GST_BIN (data->sink), "testsource"); + gst_app_src_push_buffer (GST_APP_SRC (source), app_buffer); +} + +/* called when we get a GstMessage from the source pipeline when we get EOS, we + * notify the appsrc of it. */ +static gboolean +on_source_message (GstBus * bus, GstMessage * message, ProgramData * data) +{ + GstElement *source; + + switch (GST_MESSAGE_TYPE (message)) { + case GST_MESSAGE_EOS: + g_print ("The source got dry\n"); + source = gst_bin_get_by_name (GST_BIN (data->sink), "testsource"); + gst_app_src_end_of_stream (GST_APP_SRC (source)); + break; + case GST_MESSAGE_ERROR: + g_print ("Received error\n"); + g_main_loop_quit (data->loop); + break; + default: + break; + } + return TRUE; +} + +/* called when we get a GstMessage from the sink pipeline when we get EOS, we + * exit the mainloop and this testapp. */ +static gboolean +on_sink_message (GstBus * bus, GstMessage * message, ProgramData * data) +{ + /* nil */ + switch (GST_MESSAGE_TYPE (message)) { + case GST_MESSAGE_EOS: + g_print ("Finished playback\n"); + g_main_loop_quit (data->loop); + break; + case GST_MESSAGE_ERROR: + g_print ("Received error\n"); + g_main_loop_quit (data->loop); + break; + default: + break; + } + return TRUE; +} + +int +main (int argc, char *argv[]) +{ + gchar *filename = NULL; + ProgramData *data = NULL; + gchar *string = NULL; + GstBus *bus = NULL; + GstElement *testsink = NULL; + GstElement *testsource = NULL; + + gst_init (&argc, &argv); + + if (argc == 2) + filename = g_strdup (argv[1]); + else + filename = g_strdup ("/usr/share/sounds/ekiga/ring.wav"); + + data = g_new0 (ProgramData, 1); + + data->loop = g_main_loop_new (NULL, FALSE); + + /* setting up source pipeline, we read from a file and convert to our desired + * caps. */ + string = + g_strdup_printf + ("filesrc location=\"%s\" ! wavparse ! audioconvert ! audioresample ! appsink caps=\"%s\" name=testsink", + filename, audio_caps); + g_free (filename); + data->source = gst_parse_launch (string, NULL); + g_free (string); + + if (data->source == NULL) { + g_print ("Bad source\n"); + return -1; + } + + /* to be notified of messages from this pipeline, mostly EOS */ + bus = gst_element_get_bus (data->source); + gst_bus_add_watch (bus, (GstBusFunc) on_source_message, data); + gst_object_unref (bus); + + /* we use appsink in push mode, it sends us a signal when data is available + * and we pull out the data in the signal callback. We want the appsink to + * push as fast as it can, hence the sync=false */ + testsink = gst_bin_get_by_name (GST_BIN (data->source), "testsink"); + g_object_set (G_OBJECT (testsink), "emit-signals", TRUE, "sync", FALSE, NULL); + g_signal_connect (testsink, "new-buffer", + G_CALLBACK (on_new_buffer_from_source), data); + gst_object_unref (testsink); + + /* setting up sink pipeline, we push audio data into this pipeline that will + * then play it back using the default audio sink. We have no blocking + * behaviour on the src which means that we will push the entire file into + * memory. */ + string = + g_strdup_printf ("appsrc name=testsource caps=\"%s\" ! autoaudiosink", + audio_caps); + data->sink = gst_parse_launch (string, NULL); + g_free (string); + + if (data->sink == NULL) { + g_print ("Bad sink\n"); + return -1; + } + + testsource = gst_bin_get_by_name (GST_BIN (data->sink), "testsource"); + /* configure for time-based format */ + g_object_set (testsource, "format", GST_FORMAT_TIME, NULL); + /* uncomment the next line to block when appsrc has buffered enough */ + /* g_object_set (testsource, "block", TRUE, NULL); */ + gst_object_unref (testsource); + + bus = gst_element_get_bus (data->sink); + gst_bus_add_watch (bus, (GstBusFunc) on_sink_message, data); + gst_object_unref (bus); + + /* launching things */ + gst_element_set_state (data->sink, GST_STATE_PLAYING); + gst_element_set_state (data->source, GST_STATE_PLAYING); + + /* let's run !, this loop will quit when the sink pipeline goes EOS or when an + * error occurs in the source or sink pipelines. */ + g_print ("Let's run!\n"); + g_main_loop_run (data->loop); + g_print ("Going out\n"); + + gst_element_set_state (data->source, GST_STATE_NULL); + gst_element_set_state (data->sink, GST_STATE_NULL); + + gst_object_unref (data->source); + gst_object_unref (data->sink); + g_main_loop_unref (data->loop); + g_free (data); + + return 0; +} diff --git a/gst-libs/gst/app/gstappsrc.c b/gst-libs/gst/app/gstappsrc.c index 74947af0..c98eac51 100644 --- a/gst-libs/gst/app/gstappsrc.c +++ b/gst-libs/gst/app/gstappsrc.c @@ -57,6 +57,7 @@ enum #define DEFAULT_PROP_STREAM_TYPE GST_APP_STREAM_TYPE_STREAM #define DEFAULT_PROP_MAX_BYTES 200000 #define DEFAULT_PROP_FORMAT GST_FORMAT_BYTES +#define DEFAULT_PROP_BLOCK FALSE enum { @@ -66,6 +67,7 @@ enum PROP_STREAM_TYPE, PROP_MAX_BYTES, PROP_FORMAT, + PROP_BLOCK, PROP_LAST }; @@ -187,6 +189,11 @@ gst_app_src_class_init (GstAppSrcClass * klass) 0, G_MAXUINT64, DEFAULT_PROP_MAX_BYTES, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); + g_object_class_install_property (gobject_class, PROP_BLOCK, + g_param_spec_boolean ("block", "Block", + "Block push-buffer when max-bytes are queued", + DEFAULT_PROP_BLOCK, G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS)); + /** * GstAppSrc::need-data: * @appsrc: the appsrc element that emited the signal @@ -287,6 +294,7 @@ gst_app_src_init (GstAppSrc * appsrc, GstAppSrcClass * klass) appsrc->stream_type = DEFAULT_PROP_STREAM_TYPE; appsrc->max_bytes = DEFAULT_PROP_MAX_BYTES; appsrc->format = DEFAULT_PROP_FORMAT; + appsrc->block = DEFAULT_PROP_BLOCK; } static void @@ -346,6 +354,9 @@ gst_app_src_set_property (GObject * object, guint prop_id, case PROP_FORMAT: appsrc->format = g_value_get_enum (value); break; + case PROP_BLOCK: + appsrc->block = g_value_get_boolean (value); + break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); break; @@ -382,6 +393,9 @@ gst_app_src_get_property (GObject * object, guint prop_id, GValue * value, case PROP_FORMAT: g_value_set_enum (value, appsrc->format); break; + case PROP_BLOCK: + g_value_set_boolean (value, appsrc->block); + break; default: G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec); break; @@ -396,7 +410,7 @@ gst_app_src_unlock (GstBaseSrc * bsrc) g_mutex_lock (appsrc->mutex); GST_DEBUG_OBJECT (appsrc, "unlock start"); appsrc->flushing = TRUE; - g_cond_signal (appsrc->cond); + g_cond_broadcast (appsrc->cond); g_mutex_unlock (appsrc->mutex); return TRUE; @@ -410,7 +424,7 @@ gst_app_src_unlock_stop (GstBaseSrc * bsrc) g_mutex_lock (appsrc->mutex); GST_DEBUG_OBJECT (appsrc, "unlock stop"); appsrc->flushing = FALSE; - g_cond_signal (appsrc->cond); + g_cond_broadcast (appsrc->cond); g_mutex_unlock (appsrc->mutex); return TRUE; @@ -581,6 +595,9 @@ gst_app_src_create (GstBaseSrc * bsrc, guint64 offset, guint size, gst_buffer_set_caps (*buf, appsrc->caps); + /* signal that we removed an item */ + g_cond_broadcast (appsrc->cond); + ret = GST_FLOW_OK; break; } else { @@ -806,7 +823,7 @@ gst_app_src_set_max_bytes (GstAppSrc * appsrc, guint64 max) GST_DEBUG_OBJECT (appsrc, "setting max-bytes to %u", max); appsrc->max_bytes = max; /* signal the change */ - g_cond_signal (appsrc->cond); + g_cond_broadcast (appsrc->cond); } g_mutex_unlock (appsrc->mutex); } @@ -849,28 +866,55 @@ gst_app_src_get_max_bytes (GstAppSrc * appsrc) GstFlowReturn gst_app_src_push_buffer (GstAppSrc * appsrc, GstBuffer * buffer) { + gboolean first = TRUE; + g_return_val_if_fail (appsrc, GST_FLOW_ERROR); g_return_val_if_fail (GST_IS_APP_SRC (appsrc), GST_FLOW_ERROR); g_return_val_if_fail (GST_IS_BUFFER (buffer), GST_FLOW_ERROR); g_mutex_lock (appsrc->mutex); - /* can't accept buffers when we are flushing or EOS */ - if (appsrc->flushing) - goto flushing; - if (appsrc->is_eos) - goto eos; + while (TRUE) { + /* can't accept buffers when we are flushing or EOS */ + if (appsrc->flushing) + goto flushing; + + if (appsrc->is_eos) + goto eos; + + if (appsrc->queued_bytes >= appsrc->max_bytes) { + GST_DEBUG_OBJECT (appsrc, "queue filled (%u >= %u)", + appsrc->queued_bytes, appsrc->max_bytes); + + if (first) { + /* only signal on the first push */ + g_mutex_unlock (appsrc->mutex); + + g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_ENOUGH_DATA], 0, + NULL); + + g_mutex_lock (appsrc->mutex); + /* continue to check for flushing/eos after releasing the lock */ + first = FALSE; + continue; + } + if (appsrc->block) { + GST_DEBUG_OBJECT (appsrc, "waiting for free space"); + /* we are filled, wait until a buffer gets popped or when we + * flush. */ + g_cond_wait (appsrc->cond, appsrc->mutex); + } else { + /* no need to wait for free space, we just pump data into the queue */ + break; + } + } else + break; + } GST_DEBUG_OBJECT (appsrc, "queueing buffer %p", buffer); g_queue_push_tail (appsrc->queue, buffer); - appsrc->queued_bytes += GST_BUFFER_SIZE (buffer); - if (appsrc->queued_bytes >= appsrc->max_bytes) { - GST_DEBUG_OBJECT (appsrc, "queue filled (%u >= %u), signal enough-data", - appsrc->queued_bytes, appsrc->max_bytes); - g_signal_emit (appsrc, gst_app_src_signals[SIGNAL_ENOUGH_DATA], 0, NULL); - } - g_cond_signal (appsrc->cond); + g_cond_broadcast (appsrc->cond); g_mutex_unlock (appsrc->mutex); return GST_FLOW_OK; @@ -914,7 +958,7 @@ gst_app_src_end_of_stream (GstAppSrc * appsrc) GST_DEBUG_OBJECT (appsrc, "sending EOS"); appsrc->is_eos = TRUE; - g_cond_signal (appsrc->cond); + g_cond_broadcast (appsrc->cond); g_mutex_unlock (appsrc->mutex); return GST_FLOW_OK; diff --git a/gst-libs/gst/app/gstappsrc.h b/gst-libs/gst/app/gstappsrc.h index f8173dc9..21426712 100644 --- a/gst-libs/gst/app/gstappsrc.h +++ b/gst-libs/gst/app/gstappsrc.h @@ -71,6 +71,7 @@ struct _GstAppSrc GstAppStreamType stream_type; guint64 max_bytes; GstFormat format; + gboolean block; gboolean flushing; gboolean started; |