examples: webrtc: Add bus handling to the Android and C sendrecv examples

Without a bus, messages will just pile up and errors are not handled at
all. Also without handling the LATENCY messages the latency configured
on the pipeline will be wrong.

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

subprojects/gst-examples/webrtc/android/app/src/main/jni/webrtc.c

@@ -107,7 +107,14 @@ cleanup_and_quit_loop (WebRTC * webrtc, const gchar * msg, enum AppState state)
   }
 
   if (webrtc->pipe) {
+    GstBus *bus;
+
     gst_element_set_state (webrtc->pipe, GST_STATE_NULL);
+
+    bus = gst_pipeline_get_bus (GST_PIPELINE (webrtc->pipe));
+    gst_bus_remove_watch (bus);
+    gst_object_unref (bus);
+
     gst_object_unref (webrtc->pipe);
     webrtc->pipe = NULL;
   }
@@ -328,6 +335,45 @@ add_fec_to_offer (GstElement * webrtc)
       "fec-percentage", 25, "do-nack", FALSE, NULL);
 }
 
+static gboolean
+bus_watch_cb (GstBus * bus, GstMessage * message, gpointer user_data)
+{
+  WebRTC *webrtc = user_data;
+
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_ERROR:
+    {
+      GError *error = NULL;
+      gchar *debug = NULL;
+
+      gst_message_parse_error (message, &error, &debug);
+      cleanup_and_quit_loop (webrtc, "ERROR: error on bus", APP_STATE_ERROR);
+      g_warning ("Error on bus: %s (debug: %s)", error->message, debug);
+      g_error_free (error);
+      g_free (debug);
+      break;
+    }
+    case GST_MESSAGE_WARNING:
+    {
+      GError *error = NULL;
+      gchar *debug = NULL;
+
+      gst_message_parse_warning (message, &error, &debug);
+      g_warning ("Warning on bus: %s (debug: %s)", error->message, debug);
+      g_error_free (error);
+      g_free (debug);
+      break;
+    }
+    case GST_MESSAGE_LATENCY:
+      gst_bin_recalculate_latency (GST_BIN (webrtc->pipe));
+      break;
+    default:
+      break;
+  }
+
+  return G_SOURCE_CONTINUE;
+}
+
 #define RTP_CAPS_OPUS "application/x-rtp,media=audio,encoding-name=OPUS,payload=100"
 #define RTP_CAPS_VP8 "application/x-rtp,media=video,encoding-name=VP8,payload=101"
 
@@ -352,6 +398,10 @@ start_pipeline (WebRTC * webrtc)
     goto err;
   }
 
+  bus = gst_pipeline_get_bus (GST_PIPELINE (webrtc->pipe));
+  gst_bus_add_watch (bus, bus_watch_cb, webrtc);
+  gst_object_unref (bus);
+
   webrtc->webrtcbin = gst_bin_get_by_name (GST_BIN (webrtc->pipe), "sendrecv");
   g_assert (webrtc->webrtcbin != NULL);
   add_fec_to_offer (webrtc->webrtcbin);

subprojects/gst-examples/webrtc/sendrecv/gst/webrtc-sendrecv.c

@@ -422,6 +422,44 @@ webrtcbin_get_stats (GstElement * webrtcbin)
   return G_SOURCE_REMOVE;
 }
 
+static gboolean
+bus_watch_cb (GstBus * bus, GstMessage * message, gpointer user_data)
+{
+  GstPipeline *pipeline = user_data;
+
+  switch (GST_MESSAGE_TYPE (message)) {
+    case GST_MESSAGE_ERROR:
+    {
+      GError *error = NULL;
+      gchar *debug = NULL;
+
+      gst_message_parse_error (message, &error, &debug);
+      cleanup_and_quit_loop ("ERROR: Error on bus", APP_STATE_ERROR);
+      g_warning ("Error on bus: %s (debug: %s)", error->message, debug);
+      g_error_free (error);
+      g_free (debug);
+      break;
+    }
+    case GST_MESSAGE_WARNING:
+    {
+      GError *error = NULL;
+      gchar *debug = NULL;
+
+      gst_message_parse_warning (message, &error, &debug);
+      g_warning ("Warning on bus: %s (debug: %s)", error->message, debug);
+      g_error_free (error);
+      g_free (debug);
+      break;
+    }
+    case GST_MESSAGE_LATENCY:
+      gst_bin_recalculate_latency (GST_BIN (pipeline));
+      break;
+    default:
+      break;
+  }
+
+  return G_SOURCE_CONTINUE;
+}
 
 #define STUN_SERVER "stun://stun.l.google.com:19302"
 #define RTP_TWCC_URI "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01"
@@ -431,6 +469,7 @@ webrtcbin_get_stats (GstElement * webrtcbin)
 static gboolean
 start_pipeline (gboolean create_offer, guint opus_pt, guint vp8_pt)
 {
+  GstBus *bus;
   char *audio_desc, *video_desc;
   GstStateChangeReturn ret;
   GstWebRTCICE *custom_agent;
@@ -532,6 +571,10 @@ start_pipeline (gboolean create_offer, guint opus_pt, guint vp8_pt)
   g_signal_connect (webrtc1, "notify::ice-gathering-state",
       G_CALLBACK (on_ice_gathering_state_notify), NULL);
 
+  bus = gst_pipeline_get_bus (GST_PIPELINE (pipe1));
+  gst_bus_add_watch (bus, bus_watch_cb, pipe1);
+  gst_object_unref (bus);
+
   gst_element_set_state (pipe1, GST_STATE_READY);
 
   g_signal_emit_by_name (webrtc1, "create-data-channel", "channel", NULL,
@@ -1029,8 +1072,15 @@ main (int argc, char *argv[])
     g_main_loop_unref (loop);
 
   if (pipe1) {
+    GstBus *bus;
+
     gst_element_set_state (GST_ELEMENT (pipe1), GST_STATE_NULL);
     gst_print ("Pipeline stopped\n");
+
+    bus = gst_pipeline_get_bus (GST_PIPELINE (pipe1));
+    gst_bus_remove_watch (bus);
+    gst_object_unref (bus);
+
     gst_object_unref (pipe1);
   }