wasapideviceprovider: Add support for dynamic device add/remove

Adding IMMDeviceEnumerator::RegisterEndpointNotificationCallback
in order to support device monitoring.
On OnDeviceAdded(), OnDeviceRemoved(), and OnDefaultDeviceChanged()
callback, wasapi device provider implementation will enumerate
devices again and will notify newly added and removed device
via GstDeviceProvider API.

As a bonus point, this IMMDeviceEnumerator abstraction object
will spawn a dedicated internal COM thread, so various COM thread
related issues of WASAPI plugin can be resolved by this commit.

Fixes: https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/issues/1649
Fixes: https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/issues/1110
Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/merge_requests/2484>

1 files changed, 184 insertions, 21 deletions

diff --git a/tests/check/elements/wasapi.c b/tests/check/elements/wasapi.c
index 9132b19ad..881bb682c 100644
--- a/tests/check/elements/wasapi.c
+++ b/tests/check/elements/wasapi.c
@@ -28,14 +28,143 @@
 typedef struct
 {
   GMainLoop *loop;
+  GstElement *pipeline;
+  guint n_buffers;
+  guint restart_count;
+  GstState reuse_state;
+} SrcReuseTestData;
+
+static gboolean
+src_reuse_bus_handler (GstBus * bus, GstMessage * message,
+    SrcReuseTestData * data)
+{
+  if (GST_MESSAGE_TYPE (message) == GST_MESSAGE_ERROR) {
+    GST_ERROR ("Got error message from pipeline");
+    g_main_loop_quit (data->loop);
+  }
+
+  return TRUE;
+}
+
+static void
+start_pipeline (SrcReuseTestData * data)
+{
+  GstStateChangeReturn ret;
+
+  GST_INFO ("Start pipeline");
+  ret = gst_element_set_state (data->pipeline, GST_STATE_PLAYING);
+  fail_unless (ret != GST_STATE_CHANGE_FAILURE);
+}
+
+static gboolean
+restart_pipeline (SrcReuseTestData * data)
+{
+  data->restart_count++;
+  start_pipeline (data);
+
+  return G_SOURCE_REMOVE;
+}
+
+static gboolean
+handle_handoff (SrcReuseTestData * data)
+{
+  data->n_buffers++;
+
+  /* Restart every 10 packets */
+  if (data->n_buffers > 10) {
+    GstStateChangeReturn ret;
+    data->n_buffers = 0;
+
+    ret = gst_element_set_state (data->pipeline, data->reuse_state);
+    fail_unless (ret != GST_STATE_CHANGE_FAILURE);
+
+    if (data->restart_count < 2) {
+      GST_INFO ("Restart pipeline, current restart count %d",
+          data->restart_count);
+      g_timeout_add_seconds (1, (GSourceFunc) restart_pipeline, data);
+    } else {
+      GST_INFO ("Finish test");
+      g_main_loop_quit (data->loop);
+    }
+  }
+
+  return G_SOURCE_REMOVE;
+}
+
+static void
+on_sink_handoff (GstElement * element, GstBuffer * buffer, GstPad * pad,
+    SrcReuseTestData * data)
+{
+  g_idle_add ((GSourceFunc) handle_handoff, data);
+}
+
+static void
+wasapisrc_reuse (GstState reuse_state)
+{
+  GstBus *bus;
+  GstElement *sink;
+  SrcReuseTestData data;
+
+  memset (&data, 0, sizeof (SrcReuseTestData));
+
+  data.loop = g_main_loop_new (NULL, FALSE);
+
+  data.pipeline = gst_parse_launch ("wasapisrc provide-clock=false ! queue ! "
+      "fakesink name=sink async=false", NULL);
+  fail_unless (data.pipeline != NULL);
+  data.reuse_state = reuse_state;
+
+  sink = gst_bin_get_by_name (GST_BIN (data.pipeline), "sink");
+  fail_unless (sink);
+
+  g_object_set (G_OBJECT (sink), "signal-handoffs", TRUE, NULL);
+  g_signal_connect (sink, "handoff", G_CALLBACK (on_sink_handoff), &data);
+
+  bus = gst_element_get_bus (GST_ELEMENT (data.pipeline));
+  fail_unless (bus != NULL);
+
+  gst_bus_add_watch (bus, (GstBusFunc) src_reuse_bus_handler, &data);
+  start_pipeline (&data);
+  g_main_loop_run (data.loop);
+
+  fail_unless (data.restart_count == 2);
+
+  gst_element_set_start_time (data.pipeline, GST_STATE_NULL);
+  gst_bus_remove_watch (bus);
+  gst_object_unref (bus);
+
+  gst_object_unref (data.pipeline);
+  g_main_loop_unref (data.loop);
+}
+
+/* https://gitlab.freedesktop.org/gstreamer/gst-plugins-bad/-/issues/1110 */
+GST_START_TEST (test_wasapisrc_reuse_null)
+{
+  wasapisrc_reuse (GST_STATE_NULL);
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_wasapisrc_reuse_ready)
+{
+  wasapisrc_reuse (GST_STATE_READY);
+}
+
+GST_END_TEST;
+
+typedef struct
+{
+  GMainLoop *loop;
   GstElement *pipe;
   guint rem_st_changes;
+  GstState reuse_state;
 } SinkPlayReadyTData;
 
 static gboolean
-bus_watch_cb (GstBus * bus, GstMessage * message, gpointer user_data)
+sink_reuse_bus_watch_cb (GstBus * bus, GstMessage * message, gpointer user_data)
 {
   fail_unless (message->type != GST_MESSAGE_ERROR);
+
   return G_SOURCE_CONTINUE;
 }
 
@@ -44,7 +173,7 @@ state_timer_cb (gpointer user_data)
 {
   SinkPlayReadyTData *tdata = user_data;
   GstState nxt_st = tdata->rem_st_changes % 2 == 1 ?
-      GST_STATE_READY : GST_STATE_PLAYING;
+      tdata->reuse_state : GST_STATE_PLAYING;
 
   ASSERT_SET_STATE (tdata->pipe, nxt_st, GST_STATE_CHANGE_SUCCESS);
   tdata->rem_st_changes--;
@@ -58,7 +187,8 @@ state_timer_cb (gpointer user_data)
 
 /* Test that the wasapisink can survive the state change
  * from PLAYING to READY and then back to PLAYING */
-GST_START_TEST (test_sink_play_ready)
+static void
+wasapisink_reuse (GstState reuse_state)
 {
   SinkPlayReadyTData tdata;
   GstBus *bus;
@@ -67,7 +197,9 @@ GST_START_TEST (test_sink_play_ready)
   fail_unless (tdata.pipe != NULL);
   bus = gst_element_get_bus (tdata.pipe);
   fail_unless (bus != NULL);
-  gst_bus_add_watch (bus, bus_watch_cb, NULL);
+  gst_bus_add_watch (bus, sink_reuse_bus_watch_cb, NULL);
+
+  tdata.reuse_state = reuse_state;
 
   ASSERT_SET_STATE (tdata.pipe, GST_STATE_PLAYING, GST_STATE_CHANGE_SUCCESS);
   tdata.rem_st_changes = 3;     /* -> READY -> PLAYING -> QUIT */
@@ -82,47 +214,78 @@ GST_START_TEST (test_sink_play_ready)
   gst_object_unref (tdata.pipe);
 }
 
+GST_START_TEST (test_wasapisink_reuse_null)
+{
+  wasapisink_reuse (GST_STATE_NULL);
+}
+
+GST_END_TEST;
+
+GST_START_TEST (test_wasapisink_reuse_ready)
+{
+  wasapisink_reuse (GST_STATE_READY);
+}
+
 GST_END_TEST;
 
 static gboolean
-device_is_available (const gchar * factory_name)
+check_wasapi_element (gboolean is_src)
 {
+  gboolean ret = TRUE;
   GstElement *elem;
-  gboolean avail;
+  const gchar *elem_name;
+
+  if (is_src)
+    elem_name = "wasapisrc";
+  else
+    elem_name = "wasapisink";
 
-  elem = gst_element_factory_make (factory_name, NULL);
-  if (elem == NULL) {
-    GST_INFO ("%s: not available", factory_name);
+  elem = gst_element_factory_make (elem_name, NULL);
+  if (!elem) {
+    GST_INFO ("%s is not available", elem_name);
     return FALSE;
   }
 
-  avail = gst_element_set_state (elem, GST_STATE_READY)
-      == GST_STATE_CHANGE_SUCCESS;
-  if (!avail) {
-    GST_INFO ("%s: cannot change state to ready", factory_name);
+  /* GST_STATE_READY is meaning that camera is available */
+  if (gst_element_set_state (elem, GST_STATE_READY) != GST_STATE_CHANGE_SUCCESS) {
+    GST_INFO ("cannot open device");
+    ret = FALSE;
   }
 
   gst_element_set_state (elem, GST_STATE_NULL);
   gst_object_unref (elem);
 
-  return avail;
+  return ret;
 }
 
 static Suite *
 wasapi_suite (void)
 {
-  Suite *suite = suite_create ("wasapi");
-  TCase *tc_sink = tcase_create ("sink");
+  Suite *s = suite_create ("wasapi");
+  TCase *tc_basic = tcase_create ("general");
+  gboolean have_src = FALSE;
+  gboolean have_sink = FALSE;
 
-  suite_add_tcase (suite, tc_sink);
+  suite_add_tcase (s, tc_basic);
 
-  if (device_is_available ("wasapisink")) {
-    tcase_add_test (tc_sink, test_sink_play_ready);
+  have_src = check_wasapi_element (TRUE);
+  have_sink = check_wasapi_element (FALSE);
+
+  if (!have_src && !have_sink) {
+    GST_INFO ("Skipping tests, wasapisrc/wasapisink are unavailable");
   } else {
-    GST_INFO ("Sink not available, skipping sink tests");
+    if (have_src) {
+      tcase_add_test (tc_basic, test_wasapisrc_reuse_null);
+      tcase_add_test (tc_basic, test_wasapisrc_reuse_ready);
+    }
+
+    if (have_sink) {
+      tcase_add_test (tc_basic, test_wasapisink_reuse_null);
+      tcase_add_test (tc_basic, test_wasapisink_reuse_ready);
+    }
   }
 
-  return suite;
+  return s;
 }
 
 GST_CHECK_MAIN (wasapi)