examples/app/: Add beefed up example app from bug #413418. It now also uses appsink instead of fakesink for more ulti...

Original commit message from CVS:
* 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.

6 files changed, 280 insertions, 17 deletions

diff --git a/ChangeLog b/ChangeLog
index 6b58ed1a4..ffeb1d3eb 100644
--- a/ChangeLog
+++ b/ChangeLog
@@ -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 eeae0e65e..8a2c76152 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 fc4b9419d..4f3df7779 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 000000000..6472af30a
--- /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 74947af0a..c98eac515 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 f8173dc90..214267125 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;