query: add a new bitrate query

Allows determining from downstream what the expected bitrate of a stream
may be which is useful in queue2 for setting time based limits when
upstream does not provide timing information.

Implement bitrate query handling in queue2

gst-plugins-base#60

docs/gst/gstreamer-sections.txt

@@ -2658,6 +2658,10 @@ gst_query_new_context
 gst_query_set_context
 gst_query_parse_context
 gst_query_parse_context_type
+
+gst_query_new_bitrate
+gst_query_set_bitrate
+gst_query_parse_bitrate
 <SUBSECTION Standard>
 GstQueryClass
 GST_QUERY

gst/gstpad.c

@@ -3415,6 +3415,10 @@ gst_pad_query_default (GstPad * pad, GstObject * parent, GstQuery * query)
       ret = gst_pad_query_latency_default (pad, query);
       forward = FALSE;
       break;
+    case GST_QUERY_BITRATE:
+      /* FIXME: better default handling */
+      forward = TRUE;
+      break;
     case GST_QUERY_POSITION:
     case GST_QUERY_SEEKING:
     case GST_QUERY_FORMATS:

gst/gstquark.c

@@ -75,7 +75,7 @@ static const gchar *_quark_strings[] = {
   "GstMessageStreamCollection", "collection", "stream", "stream-collection",
   "GstMessageStreamsSelected", "GstMessageRedirect", "redirect-entry-locations",
   "redirect-entry-taglists", "redirect-entry-structures",
-  "GstEventStreamGroupDone"
+  "GstEventStreamGroupDone", "GstQueryBitrate", "nominal-bitrate"
 };
 
 GQuark _priv_gst_quark_table[GST_QUARK_MAX];

gst/gstquark.h

@@ -217,7 +217,9 @@ typedef enum _GstQuarkId
   GST_QUARK_REDIRECT_ENTRY_TAGLISTS = 186,
   GST_QUARK_REDIRECT_ENTRY_STRUCTURES = 187,
   GST_QUARK_EVENT_STREAM_GROUP_DONE = 188,
-  GST_QUARK_MAX = 189
+  GST_QUARK_QUERY_BITRATE = 189,
+  GST_QUARK_NOMINAL_BITRATE = 190,
+  GST_QUARK_MAX = 191
 } GstQuarkId;
 
 extern GQuark _priv_gst_quark_table[GST_QUARK_MAX];

gst/gstquery.c

@@ -105,6 +105,7 @@ static GstQueryQuarks query_quarks[] = {
   {GST_QUERY_CAPS, "caps", 0},
   {GST_QUERY_DRAIN, "drain", 0},
   {GST_QUERY_CONTEXT, "context", 0},
+  {GST_QUERY_BITRATE, "bitrate", 0},
 
   {0, NULL, 0}
 };
@@ -2654,3 +2655,75 @@ gst_query_parse_context_type (GstQuery * query, const gchar ** context_type)
 
   return TRUE;
 }
+
+/**
+ * gst_query_new_bitrate:
+ *
+ * Constructs a new query object for querying the bitrate.
+ *
+ * Free-function: gst_query_unref()
+ *
+ * Returns: (transfer full): a new #GstQuery
+ *
+ * Since: 1.16
+ */
+GstQuery *
+gst_query_new_bitrate (void)
+{
+  GstQuery *query;
+  GstStructure *structure;
+
+  structure = gst_structure_new_id_empty (GST_QUARK (QUERY_BITRATE));
+  query = gst_query_new_custom (GST_QUERY_BITRATE, structure);
+
+  return query;
+}
+
+/**
+ * gst_query_set_bitrate:
+ * @query: a GST_QUERY_BITRATE type #GstQuery
+ * @nominal_bitrate: the nominal bitrate in bits per second
+ *
+ * Set the results of a bitrate query.  The nominal bitrate is the average
+ * bitrate expected over the length of the stream as advertised in file
+ * headers (or similar).
+ *
+ * Since: 1.16
+ */
+void
+gst_query_set_bitrate (GstQuery * query, guint nominal_bitrate)
+{
+  GstStructure *s;
+
+  g_return_if_fail (GST_QUERY_TYPE (query) == GST_QUERY_BITRATE);
+
+  s = GST_QUERY_STRUCTURE (query);
+
+  gst_structure_id_set (s,
+      GST_QUARK (NOMINAL_BITRATE), G_TYPE_UINT, nominal_bitrate, NULL);
+}
+
+/**
+ * gst_query_parse_bitrate:
+ * @query: a GST_QUERY_BITRATE type #GstQuery
+ * @nominal_bitrate: (out) (allow-none): The resulting bitrate in bits per second
+ *
+ * Get the results of a bitrate query. See also gst_query_set_bitrate().
+ *
+ * Since: 1.16
+ */
+void
+gst_query_parse_bitrate (GstQuery * query, guint * nominal_bitrate)
+{
+  GstStructure *structure;
+  const GValue *value;
+
+  g_return_if_fail (GST_QUERY_TYPE (query) == GST_QUERY_BITRATE);
+
+  structure = GST_QUERY_STRUCTURE (query);
+
+  if (nominal_bitrate) {
+    value = gst_structure_id_get_value (structure, GST_QUARK (NOMINAL_BITRATE));
+    *nominal_bitrate = g_value_get_uint (value);
+  }
+}

gst/gstquery.h

@@ -99,6 +99,7 @@ typedef enum {
  * @GST_QUERY_DRAIN: wait till all serialized data is consumed downstream
  * @GST_QUERY_CONTEXT: query the pipeline-local context from
  *     downstream or upstream (since 1.2)
+ * @GST_QUERY_BITRATE: the bitrate query (since 1.16)
  *
  * Standard predefined Query types
  */
@@ -123,7 +124,8 @@ typedef enum {
   GST_QUERY_ACCEPT_CAPS  = GST_QUERY_MAKE_TYPE (160, FLAG(BOTH)),
   GST_QUERY_CAPS         = GST_QUERY_MAKE_TYPE (170, FLAG(BOTH)),
   GST_QUERY_DRAIN        = GST_QUERY_MAKE_TYPE (180, FLAG(DOWNSTREAM) | FLAG(SERIALIZED)),
-  GST_QUERY_CONTEXT      = GST_QUERY_MAKE_TYPE (190, FLAG(BOTH))
+  GST_QUERY_CONTEXT      = GST_QUERY_MAKE_TYPE (190, FLAG(BOTH)),
+  GST_QUERY_BITRATE      = GST_QUERY_MAKE_TYPE (200, FLAG(DOWNSTREAM)),
 } GstQueryType;
 #undef FLAG
 
@@ -686,6 +688,17 @@ void            gst_query_set_context              (GstQuery *query, GstContext
 GST_API
 void            gst_query_parse_context            (GstQuery *query, GstContext **context);
 
+/* bitrate query */
+
+GST_API
+GstQuery *      gst_query_new_bitrate              (void) G_GNUC_MALLOC;
+
+GST_API
+void            gst_query_set_bitrate              (GstQuery * query, guint nominal_bitrate);
+
+GST_API
+void            gst_query_parse_bitrate            (GstQuery * query, guint * nominal_bitrate);
+
 #ifdef G_DEFINE_AUTOPTR_CLEANUP_FUNC
 G_DEFINE_AUTOPTR_CLEANUP_FUNC(GstQuery, gst_query_unref)
 #endif

plugins/elements/gstqueue2.h

@@ -109,8 +109,10 @@ struct _GstQueue2
   GstQueue2Size max_level;       /* max. amount of data allowed in the queue */
   gboolean use_buffering;
   gboolean use_tags_bitrate;
+  gboolean use_bitrate_query;
   gboolean use_rate_estimate;
   GstClockTime buffering_interval;
+  guint downstream_bitrate;     /* the bitrate reported by downstream */
 
   /* low/high watermarks for buffering */
   gint low_watermark;

tests/check/elements/queue2.c

@@ -246,9 +246,16 @@ pad_push_datablock_thread (gpointer data)
 }
 
 static GstPadProbeReturn
-block_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
+block_without_queries_probe (GstPad * pad, GstPadProbeInfo * info,
+    gpointer user_data)
 {
-  return GST_PAD_PROBE_OK;
+  GstPadProbeReturn ret = GST_PAD_PROBE_OK;
+
+  /* allows queries to pass through */
+  if ((GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_QUERY_BOTH) != 0)
+    ret = GST_PAD_PROBE_PASS;
+
+  return ret;
 }
 
 #define CHECK_FOR_BUFFERING_MSG(PIPELINE, EXPECTED_PERC) \
@@ -298,8 +305,8 @@ GST_START_TEST (test_watermark_and_fill_level)
   /* Block fakesink sinkpad flow to ensure the queue isn't emptied
    * by the prerolling sink */
   sinkpad = gst_element_get_static_pad (fakesink, "sink");
-  gst_pad_add_probe (sinkpad, GST_PAD_PROBE_TYPE_BLOCK, block_probe, NULL,
-      NULL);
+  gst_pad_add_probe (sinkpad, GST_PAD_PROBE_TYPE_BLOCK,
+      block_without_queries_probe, NULL, NULL);
   gst_object_unref (sinkpad);
 
   g_object_set (queue2,
@@ -544,6 +551,116 @@ GST_START_TEST (test_small_ring_buffer)
 
 GST_END_TEST;
 
+#define DOWNSTREAM_BITRATE (8 * 100 * 1000)
+
+static GstPadProbeReturn
+bitrate_query_probe (GstPad * pad, GstPadProbeInfo * info, gpointer user_data)
+{
+  GstPadProbeReturn ret = GST_PAD_PROBE_OK;
+
+  /* allows queries to pass through */
+  if ((GST_PAD_PROBE_INFO_TYPE (info) & GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM) !=
+      0) {
+    GstQuery *query = GST_PAD_PROBE_INFO_QUERY (info);
+
+    switch (GST_QUERY_TYPE (query)) {
+      case GST_QUERY_BITRATE:{
+        gst_query_set_bitrate (query, DOWNSTREAM_BITRATE);
+        ret = GST_PAD_PROBE_HANDLED;
+        break;
+      }
+      default:
+        break;
+    }
+  }
+
+  return ret;
+}
+
+GST_START_TEST (test_bitrate_query)
+{
+  /* This test checks the behavior of the bitrate query usage with the
+   * fill levels and buffering messages */
+
+  GstElement *pipe;
+  GstElement *queue2, *fakesink;
+  GstPad *inputpad;
+  GstPad *queue2_sinkpad;
+  GstPad *sinkpad;
+  GstSegment segment;
+  GThread *thread;
+
+  /* Setup test pipeline with one queue2 and one fakesink */
+
+  pipe = gst_pipeline_new ("pipeline");
+  queue2 = gst_element_factory_make ("queue2", NULL);
+  fail_unless (queue2 != NULL);
+  gst_bin_add (GST_BIN (pipe), queue2);
+
+  fakesink = gst_element_factory_make ("fakesink", NULL);
+  fail_unless (fakesink != NULL);
+  gst_bin_add (GST_BIN (pipe), fakesink);
+
+  /* Block fakesink sinkpad flow to ensure the queue isn't emptied
+   * by the prerolling sink */
+  sinkpad = gst_element_get_static_pad (fakesink, "sink");
+  gst_pad_add_probe (sinkpad, GST_PAD_PROBE_TYPE_BLOCK,
+      block_without_queries_probe, NULL, NULL);
+  gst_pad_add_probe (sinkpad, GST_PAD_PROBE_TYPE_QUERY_DOWNSTREAM,
+      bitrate_query_probe, NULL, NULL);
+  gst_object_unref (sinkpad);
+
+  g_object_set (queue2,
+      "use-buffering", (gboolean) TRUE,
+      "use-bitrate-query", (gboolean) TRUE,
+      "max-size-bytes", (guint) 0,
+      "max-size-buffers", (guint) 0,
+      "max-size-time", (guint64) 1 * GST_SECOND, NULL);
+
+  gst_segment_init (&segment, GST_FORMAT_TIME);
+
+  inputpad = gst_pad_new ("dummysrc", GST_PAD_SRC);
+  gst_pad_set_query_function (inputpad, queue2_dummypad_query);
+
+  queue2_sinkpad = gst_element_get_static_pad (queue2, "sink");
+  fail_unless (queue2_sinkpad != NULL);
+  fail_unless (gst_pad_link (inputpad, queue2_sinkpad) == GST_PAD_LINK_OK);
+
+  gst_pad_set_active (inputpad, TRUE);
+
+  gst_pad_push_event (inputpad, gst_event_new_stream_start ("test"));
+  gst_pad_push_event (inputpad, gst_event_new_segment (&segment));
+
+  gst_object_unref (queue2_sinkpad);
+
+  fail_unless (gst_element_link (queue2, fakesink));
+
+  /* Start pipeline in paused state to ensure the sink remains
+   * in preroll mode and blocks */
+  gst_element_set_state (pipe, GST_STATE_PAUSED);
+
+  /* When the use-buffering property is set to TRUE, a buffering
+   * message is posted. Since the queue is empty at that point,
+   * the buffering message contains a value of 0%. */
+  CHECK_FOR_BUFFERING_MSG (pipe, 0);
+
+  /* Feed data. queue will be filled to 80% (80000 bytes is pushed and
+   * with a bitrate of 100 * 1000, 80000 bytes is 80% of 1 second of data as
+   * set in the max-size-time limit) */
+  thread = g_thread_new ("push1", pad_push_datablock_thread, inputpad);
+  g_thread_join (thread);
+
+  /* Check for the buffering message; it should indicate 80% fill level
+   * (Note that the percentage from the message is normalized) */
+  CHECK_FOR_BUFFERING_MSG (pipe, 80);
+
+  gst_element_set_state (pipe, GST_STATE_NULL);
+  gst_object_unref (pipe);
+  gst_object_unref (inputpad);
+}
+
+GST_END_TEST;
+
 static Suite *
 queue2_suite (void)
 {
@@ -560,6 +677,7 @@ queue2_suite (void)
   tcase_add_test (tc_chain, test_filled_read);
   tcase_add_test (tc_chain, test_percent_overflow);
   tcase_add_test (tc_chain, test_small_ring_buffer);
+  tcase_add_test (tc_chain, test_bitrate_query);
 
   return s;
 }