GstPad: Add gst_pad_set_blocked_async_full

This allows connecting a GDestroyNotify for when the callback is removed/replaced.
Partially fixes #514717

docs/gst/gstreamer-sections.txt

@@ -1269,6 +1269,7 @@ gst_pad_is_active
 
 gst_pad_set_blocked
 gst_pad_set_blocked_async
+gst_pad_set_blocked_async_full
 GstPadBlockCallback
 gst_pad_is_blocked
 gst_pad_is_blocking

gst/gstpad.c

@@ -392,6 +392,11 @@ gst_pad_dispose (GObject * object)
 
   gst_pad_set_pad_template (pad, NULL);
 
+  if (pad->block_destroy_data && pad->block_data) {
+    pad->block_destroy_data (pad->block_data);
+    pad->block_data = NULL;
+  }
+
   G_OBJECT_CLASS (parent_class)->dispose (object);
 }
 
@@ -956,6 +961,101 @@ gst_pad_is_active (GstPad * pad)
   return result;
 }
 
+/**
+ * gst_pad_set_blocked_async_full:
+ * @pad: the #GstPad to block or unblock
+ * @blocked: boolean indicating whether the pad should be blocked or unblocked
+ * @callback: #GstPadBlockCallback that will be called when the
+ *            operation succeeds
+ * @user_data: user data passed to the callback
+ * @destroy_data: #GDestroyNotify for user_data
+ *
+ * Blocks or unblocks the dataflow on a pad. The provided callback
+ * is called when the operation succeeds; this happens right before the next
+ * attempt at pushing a buffer on the pad.
+ *
+ * This can take a while as the pad can only become blocked when real dataflow
+ * is happening.
+ * When the pipeline is stalled, for example in PAUSED, this can
+ * take an indeterminate amount of time.
+ * You can pass NULL as the callback to make this call block. Be careful with
+ * this blocking call as it might not return for reasons stated above.
+ *
+ * Returns: TRUE if the pad could be blocked. This function can fail if the
+ * wrong parameters were passed or the pad was already in the requested state.
+ *
+ * MT safe.
+ *
+ * Since: 0.10.23
+ */
+gboolean
+gst_pad_set_blocked_async_full (GstPad * pad, gboolean blocked,
+    GstPadBlockCallback callback, gpointer user_data,
+    GDestroyNotify destroy_data)
+{
+  gboolean was_blocked = FALSE;
+
+  g_return_val_if_fail (GST_IS_PAD (pad), FALSE);
+
+  GST_OBJECT_LOCK (pad);
+
+  was_blocked = GST_PAD_IS_BLOCKED (pad);
+
+  if (G_UNLIKELY (was_blocked == blocked))
+    goto had_right_state;
+
+  if (blocked) {
+    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "blocking pad");
+
+    GST_OBJECT_FLAG_SET (pad, GST_PAD_BLOCKED);
+
+    if (pad->block_destroy_data && pad->block_data &&
+        pad->block_data != user_data)
+      pad->block_destroy_data (pad->block_data);
+
+    pad->block_callback = callback;
+    pad->block_data = user_data;
+    pad->block_destroy_data = destroy_data;
+    if (!callback) {
+      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "waiting for block");
+      GST_PAD_BLOCK_WAIT (pad);
+      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "blocked");
+    }
+  } else {
+    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "unblocking pad");
+
+    GST_OBJECT_FLAG_UNSET (pad, GST_PAD_BLOCKED);
+
+    if (pad->block_destroy_data && pad->block_data &&
+        pad->block_data != user_data)
+      pad->block_destroy_data (pad->block_data);
+
+    pad->block_callback = callback;
+    pad->block_data = user_data;
+    pad->block_destroy_data = destroy_data;
+
+    GST_PAD_BLOCK_BROADCAST (pad);
+    if (!callback) {
+      /* no callback, wait for the unblock to happen */
+      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "waiting for unblock");
+      GST_PAD_BLOCK_WAIT (pad);
+      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "unblocked");
+    }
+  }
+  GST_OBJECT_UNLOCK (pad);
+
+  return TRUE;
+
+had_right_state:
+  {
+    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad,
+        "pad was in right state (%d)", was_blocked);
+    GST_OBJECT_UNLOCK (pad);
+
+    return FALSE;
+  }
+}
+
 /**
  * gst_pad_set_blocked_async:
  * @pad: the #GstPad to block or unblock
@@ -984,56 +1084,8 @@ gboolean
 gst_pad_set_blocked_async (GstPad * pad, gboolean blocked,
     GstPadBlockCallback callback, gpointer user_data)
 {
-  gboolean was_blocked = FALSE;
+  return gst_pad_set_blocked_async_full (pad, blocked,
-
+      callback, user_data, NULL);
-  g_return_val_if_fail (GST_IS_PAD (pad), FALSE);
-
-  GST_OBJECT_LOCK (pad);
-
-  was_blocked = GST_PAD_IS_BLOCKED (pad);
-
-  if (G_UNLIKELY (was_blocked == blocked))
-    goto had_right_state;
-
-  if (blocked) {
-    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "blocking pad");
-
-    GST_OBJECT_FLAG_SET (pad, GST_PAD_BLOCKED);
-    pad->block_callback = callback;
-    pad->block_data = user_data;
-    if (!callback) {
-      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "waiting for block");
-      GST_PAD_BLOCK_WAIT (pad);
-      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "blocked");
-    }
-  } else {
-    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "unblocking pad");
-
-    GST_OBJECT_FLAG_UNSET (pad, GST_PAD_BLOCKED);
-
-    pad->block_callback = callback;
-    pad->block_data = user_data;
-
-    GST_PAD_BLOCK_BROADCAST (pad);
-    if (!callback) {
-      /* no callback, wait for the unblock to happen */
-      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "waiting for unblock");
-      GST_PAD_BLOCK_WAIT (pad);
-      GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad, "unblocked");
-    }
-  }
-  GST_OBJECT_UNLOCK (pad);
-
-  return TRUE;
-
-had_right_state:
-  {
-    GST_CAT_LOG_OBJECT (GST_CAT_SCHEDULING, pad,
-        "pad was in right state (%d)", was_blocked);
-    GST_OBJECT_UNLOCK (pad);
-
-    return FALSE;
-  }
 }
 
 /**

gst/gstpad.h

@@ -629,8 +629,11 @@ struct _GstPad {
   /* iterate internal links */
   GstPadIterIntLinkFunction     iterintlinkfunc;
 
+  /* free block_data */
+  GDestroyNotify block_destroy_data;
+
   /*< private >*/
-  gpointer _gst_reserved[GST_PADDING - 1];
+  gpointer _gst_reserved[GST_PADDING - 2];
 };
 
 struct _GstPadClass {
@@ -809,6 +812,9 @@ gboolean		gst_pad_activate_push			(GstPad *pad, gboolean active);
 gboolean		gst_pad_set_blocked			(GstPad *pad, gboolean blocked);
 gboolean		gst_pad_set_blocked_async		(GstPad *pad, gboolean blocked,
 								 GstPadBlockCallback callback, gpointer user_data);
+gboolean		gst_pad_set_blocked_async_full		(GstPad *pad, gboolean blocked,
+								 GstPadBlockCallback callback, gpointer user_data,
+                 GDestroyNotify destroy_data);
 gboolean		gst_pad_is_blocked			(GstPad *pad);
 gboolean		gst_pad_is_blocking			(GstPad *pad);
 

tests/check/gst/gstpad.c

@@ -624,6 +624,191 @@ GST_START_TEST (test_get_caps_must_be_copy)
 
 GST_END_TEST;
 
+static void
+unblock_async_cb (GstPad * pad, gboolean blocked, gpointer user_data)
+{
+  gboolean *bool_user_data = (gboolean *) user_data;
+
+  /* here we should have blocked == 1 unblocked == 0 */
+  fail_unless (bool_user_data[0] == TRUE);
+  fail_unless (bool_user_data[1] == FALSE);
+
+  bool_user_data[1] = TRUE;
+}
+
+static void
+block_async_cb (GstPad * pad, gboolean blocked, gpointer user_data)
+{
+  gboolean *bool_user_data = (gboolean *) user_data;
+
+  /* here we should have blocked == 0 unblocked == 0 */
+  fail_unless (bool_user_data[0] == FALSE);
+  fail_unless (bool_user_data[1] == FALSE);
+
+  bool_user_data[0] = blocked;
+
+  gst_pad_set_blocked_async (pad, FALSE, unblock_async_cb, user_data);
+}
+
+GST_START_TEST (test_block_async)
+{
+  GstPad *pad;
+  /* we set data[0] = TRUE when the pad is blocked, data[1] = TRUE when it's
+   * unblocked */
+  gboolean data[2] = { FALSE, FALSE };
+
+  pad = gst_pad_new ("src", GST_PAD_SRC);
+  fail_unless (pad != NULL);
+
+  gst_pad_set_active (pad, TRUE);
+  gst_pad_set_blocked_async (pad, TRUE, block_async_cb, &data);
+
+  fail_unless (data[0] == FALSE);
+  fail_unless (data[1] == FALSE);
+  gst_pad_push (pad, gst_buffer_new ());
+
+  gst_object_unref (pad);
+}
+
+GST_END_TEST;
+
+#if 0
+static void
+block_async_second (GstPad * pad, gboolean blocked, gpointer user_data)
+{
+  gst_pad_set_blocked_async (pad, FALSE, unblock_async_cb, NULL);
+}
+
+static void
+block_async_first (GstPad * pad, gboolean blocked, gpointer user_data)
+{
+  static int n_calls = 0;
+  gboolean *bool_user_data = (gboolean *) user_data;
+
+  if (++n_calls > 1)
+    /* we expect this callback to be called only once */
+    g_warn_if_reached ();
+
+  *bool_user_data = blocked;
+
+  /* replace block_async_first with block_async_second so next time the pad is
+   * blocked the latter should be called */
+  gst_pad_set_blocked_async (pad, TRUE, block_async_second, NULL);
+
+  /* unblock temporarily, in the next push block_async_second should be called
+   */
+  gst_pad_push_event (pad, gst_event_new_flush_start ());
+}
+
+GST_START_TEST (test_block_async_replace_callback)
+{
+  GstPad *pad;
+  gboolean blocked;
+
+  pad = gst_pad_new ("src", GST_PAD_SRC);
+  fail_unless (pad != NULL);
+  gst_pad_set_active (pad, TRUE);
+
+  gst_pad_set_blocked_async (pad, TRUE, block_async_first, &blocked);
+  blocked = FALSE;
+
+  gst_pad_push (pad, gst_buffer_new ());
+  fail_unless (blocked == TRUE);
+  /* block_async_first flushes to unblock */
+  gst_pad_push_event (pad, gst_event_new_flush_stop ());
+
+  /* push again, this time block_async_second should be called */
+  gst_pad_push (pad, gst_buffer_new ());
+  fail_unless (blocked == TRUE);
+
+  gst_object_unref (pad);
+}
+
+GST_END_TEST;
+#endif
+
+static void
+block_async_full_destroy (gpointer user_data)
+{
+  gint *state = (gint *) user_data;
+
+  fail_unless (*state < 2);
+
+  *state = 2;
+}
+
+static void
+block_async_full_cb (GstPad * pad, gboolean blocked, gpointer user_data)
+{
+  *(gint *) user_data = (gint) blocked;
+
+  gst_pad_push_event (pad, gst_event_new_flush_start ());
+}
+
+GST_START_TEST (test_block_async_full_destroy)
+{
+  GstPad *pad;
+  /* 0 = unblocked, 1 = blocked, 2 = destroyed */
+  gint state = 0;
+
+  pad = gst_pad_new ("src", GST_PAD_SRC);
+  fail_unless (pad != NULL);
+  gst_pad_set_active (pad, TRUE);
+
+  gst_pad_set_blocked_async_full (pad, TRUE, block_async_full_cb,
+      &state, block_async_full_destroy);
+  fail_unless (state == 0);
+
+  gst_pad_push (pad, gst_buffer_new ());
+  /* block_async_full_cb sets state to 1 and then flushes to unblock temporarily
+   */
+  fail_unless (state == 1);
+  gst_pad_push_event (pad, gst_event_new_flush_stop ());
+
+  /* call with the same user_data, should not call the destroy_notify function
+   */
+  gst_pad_set_blocked_async_full (pad, TRUE, block_async_full_cb,
+      &state, block_async_full_destroy);
+  fail_unless (state == 1);
+
+  /* now change user_data (to NULL in this case) so destroy_notify should be
+   * called */
+  gst_pad_set_blocked_async_full (pad, FALSE, block_async_full_cb,
+      NULL, block_async_full_destroy);
+  fail_unless (state == 2);
+
+  gst_object_unref (pad);
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_block_async_full_destroy_dispose)
+{
+  GstPad *pad;
+  /* 0 = unblocked, 1 = blocked, 2 = destroyed */
+  gint state = 0;
+
+  pad = gst_pad_new ("src", GST_PAD_SRC);
+  fail_unless (pad != NULL);
+  gst_pad_set_active (pad, TRUE);
+
+  gst_pad_set_blocked_async_full (pad, TRUE, block_async_full_cb,
+      &state, block_async_full_destroy);
+
+  gst_pad_push (pad, gst_buffer_new ());
+  /* block_async_full_cb sets state to 1 and then flushes to unblock temporarily
+   */
+  fail_unless_equals_int (state, 1);
+  gst_pad_push_event (pad, gst_event_new_flush_stop ());
+
+  /* gst_pad_dispose calls the destroy_notify function if necessary */
+  gst_object_unref (pad);
+
+  fail_unless_equals_int (state, 2);
+}
+
+GST_END_TEST;
+
 static Suite *
 gst_pad_suite (void)
 {
@@ -646,6 +831,12 @@ gst_pad_suite (void)
   tcase_add_test (tc_chain, test_src_unref_unlink);
   tcase_add_test (tc_chain, test_sink_unref_unlink);
   tcase_add_test (tc_chain, test_get_caps_must_be_copy);
+  tcase_add_test (tc_chain, test_block_async);
+#if 0
+  tcase_add_test (tc_chain, test_block_async_replace_callback);
+#endif
+  tcase_add_test (tc_chain, test_block_async_full_destroy);
+  tcase_add_test (tc_chain, test_block_async_full_destroy_dispose);
 
   return s;
 }

win32/common/libgstreamer.def

@@ -594,6 +594,7 @@ EXPORTS
 	gst_pad_set_active
 	gst_pad_set_blocked
 	gst_pad_set_blocked_async
+	gst_pad_set_blocked_async_full
 	gst_pad_set_bufferalloc_function
 	gst_pad_set_caps
 	gst_pad_set_chain_function