webrtc: janus: handle (stopped-)talking events

Expose those events using a signal.

Fix those errors when joining a Janus room configured with
'audiolevel_event: true'.

ERROR   webrtc-janusvr-signaller imp.rs:408:gstrswebrtc::janusvr_signaller:👿:Signaller::handle_msg:<GstJanusVRWebRTCSignaller@0x560cf2a55100> Unknown message from server: {
   "janus": "event",
   "session_id": 2384862538500481,
   "sender": 1867822625190966,
   "plugindata": {
      "plugin": "janus.plugin.videoroom",
      "data": {
         "videoroom": "talking",
         "room": 7564250471742314,
         "id": 6815475717947398,
         "mindex": 0,
         "mid": "0",
         "audio-level-dBov-avg": 37.939998626708984
      }
   }
}
ERROR   webrtc-janusvr-signaller imp.rs:408:gstrswebrtc::janusvr_signaller:👿:Signaller::handle_msg:<GstJanusVRWebRTCSignaller@0x560cf2a55100> Unknown message from server: {
   "janus": "event",
   "session_id": 2384862538500481,
   "sender": 1867822625190966,
   "plugindata": {
      "plugin": "janus.plugin.videoroom",
      "data": {
         "videoroom": "stopped-talking",
         "room": 7564250471742314,
         "id": 6815475717947398,
         "mindex": 0,
         "mid": "0",
         "audio-level-dBov-avg": 40.400001525878906
      }
   }
}

Part-of: <https://gitlab.freedesktop.org/gstreamer/gst-plugins-rs/-/merge_requests/1481>

net/webrtc/src/janusvr_signaller/imp.rs

@@ -9,7 +9,7 @@ use futures::channel::mpsc;
 use futures::sink::SinkExt;
 use futures::stream::StreamExt;
 use gst::glib;
-use gst::glib::Properties;
+use gst::glib::{subclass::Signal, Properties};
 use gst::prelude::*;
 use gst::subclass::prelude::*;
 use http::Uri;
@@ -42,7 +42,7 @@ fn transaction_id() -> String {
 #[serde(untagged)]
 /// Ids are either u64 (default) or string in Janus, depending of the
 /// `string_ids` configuration in the videoroom plugin config file.
-enum JanusId {
+pub(crate) enum JanusId {
     Str(String),
     Num(u64),
 }
@@ -180,12 +180,20 @@ struct RoomEvent {
 }
 
 #[derive(Serialize, Deserialize, Debug)]
-#[serde(tag = "videoroom")]
+struct RoomTalking {
+    room: JanusId,
+    id: JanusId,
+    #[serde(rename = "audio-level-dBov-avg")]
+    audio_level: f32,
+}
+
+#[derive(Serialize, Deserialize, Debug)]
+#[serde(tag = "videoroom", rename_all = "kebab-case")]
 enum VideoRoomData {
-    #[serde(rename = "joined")]
     Joined(RoomJoined),
-    #[serde(rename = "event")]
     Event(RoomEvent),
+    Talking(RoomTalking),
+    StoppedTalking(RoomTalking),
 }
 
 #[derive(Serialize, Deserialize, Debug)]
@@ -476,6 +484,12 @@ impl Signaller {
                                 }
                             }
                         }
+                        VideoRoomData::Talking(talking) => {
+                            self.emit_talking(true, talking.id, talking.audio_level);
+                        }
+                        VideoRoomData::StoppedTalking(talking) => {
+                            self.emit_talking(false, talking.id, talking.audio_level);
+                        }
                     }
                 }
             }
@@ -721,6 +735,11 @@ impl Signaller {
             }
         }
     }
+
+    fn emit_talking(&self, talking: bool, id: JanusId, audio_level: f32) {
+        let obj = self.obj();
+        (obj.class().as_ref().emit_talking)(&obj, talking, id, audio_level)
+    }
 }
 
 impl SignallableImpl for Signaller {
@@ -825,9 +844,33 @@ pub mod signaller_u64 {
     }
 
     #[glib::derived_properties]
-    impl ObjectImpl for SignallerU64 {}
+    impl ObjectImpl for SignallerU64 {
+        fn signals() -> &'static [Signal] {
+            static SIGNALS: Lazy<Vec<Signal>> = Lazy::new(|| {
+                vec![
+                    /**
+                     * GstJanusVRWebRTCSignallerU64::talking:
+                     * @self: A #GstJanusVRWebRTCSignallerStr
+                     * @talking: if the publisher is talking, or not
+                     * @id: unique numeric ID of the publisher
+                     * @audio-level: average value of audio level, 127=muted, 0='too loud'
+                     */
+                    Signal::builder("talking")
+                        .param_types([bool::static_type(), u64::static_type(), f32::static_type()])
+                        .build(),
+                ]
+            });
 
-    impl super::super::JanusVRSignallerImpl for SignallerU64 {}
+            SIGNALS.as_ref()
+        }
+    }
+
+    impl super::super::JanusVRSignallerImpl for SignallerU64 {
+        fn emit_talking(&self, talking: bool, id: super::JanusId, audio_level: f32) {
+            self.obj()
+                .emit_by_name::<()>("talking", &[&talking, &id.as_num(), &audio_level]);
+        }
+    }
 
     impl SignallerU64 {
         fn get_room_id(&self) -> u64 {
@@ -892,9 +935,33 @@ pub mod signaller_str {
     }
 
     #[glib::derived_properties]
-    impl ObjectImpl for SignallerStr {}
+    impl ObjectImpl for SignallerStr {
+        fn signals() -> &'static [Signal] {
+            static SIGNALS: Lazy<Vec<Signal>> = Lazy::new(|| {
+                vec![
+                    /**
+                     * GstJanusVRWebRTCSignallerStr::talking:
+                     * @self: A #GstJanusVRWebRTCSignallerStr
+                     * @talking: if the publisher is talking, or not
+                     * @id: unique string ID of the publisher
+                     * @audio-level: average value of audio level, 127=muted, 0='too loud'
+                     */
+                    Signal::builder("talking")
+                        .param_types([bool::static_type(), str::static_type(), f32::static_type()])
+                        .build(),
+                ]
+            });
 
-    impl super::super::JanusVRSignallerImpl for SignallerStr {}
+            SIGNALS.as_ref()
+        }
+    }
+
+    impl super::super::JanusVRSignallerImpl for SignallerStr {
+        fn emit_talking(&self, talking: bool, id: super::JanusId, audio_level: f32) {
+            self.obj()
+                .emit_by_name::<()>("talking", &[&talking, &id.as_string(), &audio_level]);
+        }
+    }
 
     impl SignallerStr {
         fn get_room_id(&self) -> String {

net/webrtc/src/janusvr_signaller/mod.rs

@@ -1,7 +1,7 @@
 // SPDX-License-Identifier: MPL-2.0
 
 use crate::signaller::Signallable;
-use gst::{glib, glib::subclass::prelude::*};
+use gst::{glib, glib::prelude::*, glib::subclass::prelude::*};
 
 mod imp;
 
@@ -10,11 +10,16 @@ glib::wrapper! {
     pub struct JanusVRSignaller(ObjectSubclass<imp::Signaller>) @implements Signallable;
 }
 
-trait JanusVRSignallerImpl: ObjectImpl {}
+trait JanusVRSignallerImpl: ObjectImpl {
+    fn emit_talking(&self, talking: bool, id: imp::JanusId, audio_level: f32);
+}
 
 #[repr(C)]
 pub struct JanusVRSignallerClass {
     parent: glib::object::ObjectClass,
+
+    // virtual methods
+    emit_talking: fn(&JanusVRSignaller, talking: bool, id: imp::JanusId, audio_level: f32),
 }
 
 unsafe impl ClassStruct for JanusVRSignallerClass {
@@ -32,6 +37,13 @@ impl std::ops::Deref for JanusVRSignallerClass {
 unsafe impl<T: JanusVRSignallerImpl> IsSubclassable<T> for JanusVRSignaller {
     fn class_init(class: &mut glib::Class<Self>) {
         Self::parent_class_init::<T>(class);
+
+        let class = class.as_mut();
+
+        class.emit_talking = |obj, talking, id, audio_level| unsafe {
+            let imp = obj.unsafe_cast_ref::<T::Type>().imp();
+            imp.emit_talking(talking, id, audio_level)
+        };
     }
 }