--- /dev/null
+#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;
+}
#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
{
PROP_STREAM_TYPE,
PROP_MAX_BYTES,
PROP_FORMAT,
+ PROP_BLOCK,
PROP_LAST
};
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
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
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;
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;
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;
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;
gst_buffer_set_caps (*buf, appsrc->caps);
+ /* signal that we removed an item */
+ g_cond_broadcast (appsrc->cond);
+
ret = GST_FLOW_OK;
break;
} else {
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);
}
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;
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;
projects / platform / upstream / gst-plugins-bad.git / commitdiff