queue2: Fix missing/dropped buffering messages at startup

This fixes a bug that occurs when an attempt is made to post a buffering
message before the queue2 was assigned a bus. One common situation where
this happens is when the use-buffering property is set to TRUE before the
queue2 was added to a bin.

If the result of gst_element_post_message() is not checked, and the
aforementioned situation occurs, then last_posted_buffering_percent and
percent_changed will still be updated, as if posting the message succeeded.
Later attempts to post again will not do anything because the code then
assumes that a message with the same percentage was previously posted
successfully and posting again is redundant.

Updating these variables only if posting succeed and explicitely
posting a buffering message in the READY->PAUSED state change ensure that
a buffering message is posted as early as possible.

Part-of: <https://gitlab.freedesktop.org/gstreamer/gstreamer/-/merge_requests/297>

plugins/elements/gstqueue2.c

@@ -49,6 +49,12 @@
  *
  * The temp-location property will be used to notify the application of the
  * allocated filename.
+ *
+ * If the #GstQueue2:use-buffering property is set to TRUE, and any writable
+ * property is modified, #GstQueue2 will attempt to post a buffering message
+ * if the changes to the properties also cause the buffering percentage to be
+ * changed (for example, because the queue's capacity was changed and it already
+ * contains some data).
  */
 
 #ifdef HAVE_CONFIG_H
@@ -1132,10 +1138,7 @@ gst_queue2_get_buffering_message (GstQueue2 * queue)
 
       gst_message_set_buffering_stats (msg, queue->mode, queue->avg_in,
           queue->avg_out, queue->buffering_left);
-
-      queue->last_posted_buffering_percent = percent;
     }
-    queue->percent_changed = FALSE;
   }
 
   return msg;
@@ -1151,8 +1154,20 @@ gst_queue2_post_buffering (GstQueue2 * queue)
   msg = gst_queue2_get_buffering_message (queue);
   GST_QUEUE2_MUTEX_UNLOCK (queue);
 
-  if (msg != NULL)
+  if (msg != NULL) {
-    gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
+    if (gst_element_post_message (GST_ELEMENT_CAST (queue), msg)) {
+      GST_QUEUE2_MUTEX_LOCK (queue);
+      /* Set these states only if posting the message succeeded. Otherwise,
+       * this post attempt failed, and the next one won't be done, because
+       * gst_queue2_get_buffering_message() checks these states and decides
+       * based on their values that it won't produce a message. */
+      queue->last_posted_buffering_percent = queue->percent_changed;
+      queue->percent_changed = FALSE;
+      GST_QUEUE2_MUTEX_UNLOCK (queue);
+      GST_DEBUG_OBJECT (queue, "successfully posted buffering message");
+    } else
+      GST_DEBUG_OBJECT (queue, "could not post buffering message");
+  }
 
   g_mutex_unlock (&queue->buffering_post_lock);
 }
@@ -2171,11 +2186,26 @@ gst_queue2_create_write (GstQueue2 * queue, GstBuffer * buffer)
       update_buffering (queue);
       msg = gst_queue2_get_buffering_message (queue);
       if (msg) {
+        gboolean post_ok;
+
         GST_QUEUE2_MUTEX_UNLOCK (queue);
+
         g_mutex_lock (&queue->buffering_post_lock);
-        gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
+        post_ok = gst_element_post_message (GST_ELEMENT_CAST (queue), msg);
         g_mutex_unlock (&queue->buffering_post_lock);
+
         GST_QUEUE2_MUTEX_LOCK (queue);
+
+        if (post_ok) {
+          /* Set these states only if posting the message succeeded. Otherwise,
+           * this post attempt failed, and the next one won't be done, because
+           * gst_queue2_get_buffering_message() checks these states and decides
+           * based on their values that it won't produce a message. */
+          queue->last_posted_buffering_percent = queue->percent_changed;
+          queue->percent_changed = FALSE;
+          GST_DEBUG_OBJECT (queue, "successfully posted buffering message");
+        } else
+          GST_DEBUG_OBJECT (queue, "could not post buffering message");
       }
     }
 
@@ -3750,6 +3780,14 @@ gst_queue2_change_state (GstElement * element, GstStateChange transition)
       gst_event_replace (&queue->stream_start_event, NULL);
       GST_QUEUE2_MUTEX_UNLOCK (queue);
       query_downstream_bitrate (queue);
+
+      /* Post a buffering message now to make sure the application receives
+       * a buffering message as early as possible. This prevents situations
+       * where the pipeline's state is set to PLAYING too early, before
+       * buffering actually finished. */
+      update_buffering (queue);
+      gst_queue2_post_buffering (queue);
+
       break;
     case GST_STATE_CHANGE_PAUSED_TO_PLAYING:
       break;

tests/check/elements/queue2.c

@@ -661,6 +661,58 @@ GST_START_TEST (test_bitrate_query)
 
 GST_END_TEST;
 
+GST_START_TEST (test_ready_paused_buffering_message)
+{
+  /* This test verifies that a buffering message is posted during the
+   * READY->PAUSED state change. */
+
+  GstElement *pipe;
+  GstElement *fakesrc, *queue2, *fakesink;
+
+  /* Set up simple test pipeline. */
+
+  pipe = gst_pipeline_new ("pipeline");
+
+  /* Set up the fakesrc to actually produce data. */
+  fakesrc = gst_element_factory_make ("fakesrc", NULL);
+  fail_unless (fakesrc != NULL);
+  g_object_set (G_OBJECT (fakesrc), "format", (gint) 3, "filltype", (gint) 2,
+      "sizetype", (gint) 2, "sizemax", (gint) 4096, "datarate", (gint) 4096,
+      NULL);
+
+  queue2 = gst_element_factory_make ("queue2", NULL);
+  fail_unless (queue2 != NULL);
+  /* Note that use-buffering is set _before_ the queue2 got added to pipe.
+   * This is intentional. queue2's set_property function attempts to post a
+   * buffering message. This fails silently, because without having been added
+   * to a bin, queue2 won't have been assigned a bus, so it cannot post that
+   * message anywhere. In such a case, the next attempt to post a buffering
+   * message must always actually be attempted. (Normally, queue2 performs
+   * internal checks to see whether or not the buffering message would be
+   * redundant because a prior message with the same percentage was already
+   * posted. But these checked only make sense if the previous posting attempt
+   * succeeded.) */
+  g_object_set (queue2, "use-buffering", (gboolean) TRUE, NULL);
+
+  fakesink = gst_element_factory_make ("fakesink", NULL);
+  fail_unless (fakesink != NULL);
+
+  gst_bin_add_many (GST_BIN (pipe), fakesrc, queue2, fakesink, NULL);
+
+  /* Set the pipeline to PAUSED. This should cause queue2 to attempt to post
+   * a buffering message during its READY->PAUSED state change. And this should
+   * succeed, since queue2 has been added to pipe by now. */
+  gst_element_set_state (pipe, GST_STATE_PAUSED);
+
+  /* Look for the expected 0% buffering message. */
+  CHECK_FOR_BUFFERING_MSG (pipe, 0);
+
+  gst_element_set_state (pipe, GST_STATE_NULL);
+  gst_object_unref (pipe);
+}
+
+GST_END_TEST;
+
 static Suite *
 queue2_suite (void)
 {
@@ -678,6 +730,7 @@ queue2_suite (void)
   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);
+  tcase_add_test (tc_chain, test_ready_paused_buffering_message);
 
   return s;
 }