queue: add "flush-on-eos" property

In flush-on-eos=true mode any data remaining in the queue is
discarded when an EOS event is received, and the EOS passed
downstream as soon as possible (instead of waiting for all
buffers in the queue to get processed by downstream first).

May or may not be useful in capture/encoding scenarios.

3 files changed, 46 insertions, 4 deletions

diff --git a/docs/plugins/gstreamer-plugins.args b/docs/plugins/gstreamer-plugins.args
index be3f66201..612608e73 100644
--- a/docs/plugins/gstreamer-plugins.args
+++ b/docs/plugins/gstreamer-plugins.args
@@ -599,6 +599,16 @@
 </ARG>
 
 <ARG>
+<NAME>GstQueue::flush-on-eos</NAME>
+<TYPE>gboolean</TYPE>
+<RANGE></RANGE>
+<FLAGS>rw</FLAGS>
+<NICK>Flush on EOS</NICK>
+<BLURB>Discard all data in the queue when an EOS event is received.</BLURB>
+<DEFAULT>FALSE</DEFAULT>
+</ARG>
+
+<ARG>
 <NAME>GstFileSink::location</NAME>
 <TYPE>gchar*</TYPE>
 <RANGE></RANGE>
diff --git a/plugins/elements/gstqueue.c b/plugins/elements/gstqueue.c
index 8b06fe1d6..b6dfa3600 100644
--- a/plugins/elements/gstqueue.c
+++ b/plugins/elements/gstqueue.c
@@ -119,7 +119,8 @@ enum
   PROP_MIN_THRESHOLD_BYTES,
   PROP_MIN_THRESHOLD_TIME,
   PROP_LEAKY,
-  PROP_SILENT
+  PROP_SILENT,
+  PROP_FLUSH_ON_EOS
 };
 
 /* default property values */
@@ -359,6 +360,26 @@ gst_queue_class_init (GstQueueClass * klass)
           "Don't emit queue signals", FALSE,
           G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
 
+  /**
+   * GstQueue:flush-on-eos
+   *
+   * Discard all data in the queue when an EOS event is received, and pass
+   * on the EOS event as soon as possible (instead of waiting until all
+   * buffers in the queue have been processed, which is the default behaviour).
+   *
+   * Flushing the queue on EOS might be useful when capturing and encoding
+   * from a live source, to finish up the recording quickly in cases when
+   * the encoder is slow. Note that this might mean some data from the end of
+   * the recoding data might be lost though (never more than the configured
+   * max. sizes though).
+   *
+   * Since: 1.2
+   */
+  g_object_class_install_property (gobject_class, PROP_FLUSH_ON_EOS,
+      g_param_spec_boolean ("flush-on-eos", "Flush on EOS",
+          "Discard all data in the queue when an EOS event is received", FALSE,
+          G_PARAM_READWRITE | G_PARAM_STATIC_STRINGS));
+
   gobject_class->finalize = gst_queue_finalize;
 
   gst_element_class_set_static_metadata (gstelement_class,
@@ -601,11 +622,14 @@ gst_queue_locked_enqueue_event (GstQueue * queue, gpointer item)
 
   switch (GST_EVENT_TYPE (event)) {
     case GST_EVENT_EOS:
+      GST_CAT_LOG_OBJECT (queue_dataflow, queue, "got EOS from upstream");
       /* Zero the thresholds, this makes sure the queue is completely
        * filled and we can read all data from the queue. */
-      GST_QUEUE_CLEAR_LEVEL (queue->min_threshold);
+      if (queue->flush_on_eos)
+        gst_queue_locked_flush (queue);
+      else
+        GST_QUEUE_CLEAR_LEVEL (queue->min_threshold);
       /* mark the queue as EOS. This prevents us from accepting more data. */
-      GST_CAT_LOG_OBJECT (queue_dataflow, queue, "got EOS from upstream");
       queue->eos = TRUE;
       break;
     case GST_EVENT_SEGMENT:
@@ -754,7 +778,7 @@ gst_queue_handle_sink_event (GstPad * pad, GstObject * parent, GstEvent * event)
         gst_queue_locked_enqueue_event (queue, event);
         GST_QUEUE_MUTEX_UNLOCK (queue);
       } else {
-        /* non-serialized events are passed upstream. */
+        /* non-serialized events are forwarded downstream immediately */
         gst_pad_push_event (queue->srcpad, event);
       }
       break;
@@ -1424,6 +1448,9 @@ gst_queue_set_property (GObject * object,
     case PROP_SILENT:
       queue->silent = g_value_get_boolean (value);
       break;
+    case PROP_FLUSH_ON_EOS:
+      queue->flush_on_eos = g_value_get_boolean (value);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
@@ -1474,6 +1501,9 @@ gst_queue_get_property (GObject * object,
     case PROP_SILENT:
       g_value_set_boolean (value, queue->silent);
       break;
+    case PROP_FLUSH_ON_EOS:
+      g_value_set_boolean (value, queue->flush_on_eos);
+      break;
     default:
       G_OBJECT_WARN_INVALID_PROPERTY_ID (object, prop_id, pspec);
       break;
diff --git a/plugins/elements/gstqueue.h b/plugins/elements/gstqueue.h
index a262d6190..82ba83be1 100644
--- a/plugins/elements/gstqueue.h
+++ b/plugins/elements/gstqueue.h
@@ -134,6 +134,8 @@ struct _GstQueue {
   gboolean newseg_applied_to_src;
 
   gboolean last_query;
+
+  gboolean flush_on_eos; /* flush on EOS */
 };
 
 struct _GstQueueClass {