Add support for logging RTP/RTCP packets

Cargo.lock

@@ -683,6 +683,12 @@ dependencies = [
  "cfg-if",
 ]
 
+[[package]]
+name = "ipnet"
+version = "2.4.0"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "35e70ee094dc02fd9c13fdad4940090f22dbd6ac7c9e7094a46cf0232a50bc7c"
+
 [[package]]
 name = "itertools"
 version = "0.10.3"
@@ -762,6 +768,7 @@ dependencies = [
  "rand",
  "rcgen",
  "ring",
+ "rtcp",
  "rustls",
  "rustls-native-certs",
  "serde",
@@ -1063,6 +1070,17 @@ dependencies = [
  "paste",
 ]
 
+[[package]]
+name = "parking_lot"
+version = "0.11.2"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "7d17b78036a60663b797adeaee46f5c9dfebb86948d1255007a1d6be0271ff99"
+dependencies = [
+ "instant",
+ "lock_api",
+ "parking_lot_core 0.8.5",
+]
+
 [[package]]
 name = "parking_lot"
 version = "0.12.0"
@@ -1070,7 +1088,21 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "87f5ec2493a61ac0506c0f4199f99070cbe83857b0337006a30f3e6719b8ef58"
 dependencies = [
  "lock_api",
- "parking_lot_core",
+ "parking_lot_core 0.9.1",
+]
+
+[[package]]
+name = "parking_lot_core"
+version = "0.8.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "d76e8e1493bcac0d2766c42737f34458f1c8c50c0d23bcb24ea953affb273216"
+dependencies = [
+ "cfg-if",
+ "instant",
+ "libc",
+ "redox_syscall",
+ "smallvec",
+ "winapi",
 ]
 
 [[package]]
@@ -1272,6 +1304,17 @@ dependencies = [
  "winapi",
 ]
 
+[[package]]
+name = "rtcp"
+version = "0.6.5"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "80d9625e47edb43aca711ec826ad12154d364ada9e60f4e6f8d40471b3e1e156"
+dependencies = [
+ "bytes",
+ "thiserror",
+ "webrtc-util",
+]
+
 [[package]]
 name = "rustls"
 version = "0.20.4"
@@ -1653,6 +1696,7 @@ dependencies = [
  "mio",
  "num_cpus",
  "once_cell",
+ "parking_lot 0.12.0",
  "pin-project-lite",
  "signal-hook-registry",
  "socket2",
@@ -1781,7 +1825,7 @@ version = "0.3.9"
 source = "registry+https://github.com/rust-lang/crates.io-index"
 checksum = "9e0ab7bdc962035a87fba73f3acca9b8a8d0034c2e6f60b84aeaaddddc155dce"
 dependencies = [
- "parking_lot",
+ "parking_lot 0.12.0",
  "sharded-slab",
  "smallvec",
  "thread_local",
@@ -1996,6 +2040,28 @@ dependencies = [
  "webpki",
 ]
 
+[[package]]
+name = "webrtc-util"
+version = "0.5.3"
+source = "registry+https://github.com/rust-lang/crates.io-index"
+checksum = "60b69bdea720881eddee50bd969be052ed95f04ccc5a6f684495131afb87e31c"
+dependencies = [
+ "async-trait",
+ "bitflags",
+ "bytes",
+ "cc",
+ "ipnet",
+ "lazy_static",
+ "libc",
+ "log",
+ "nix",
+ "parking_lot 0.11.2",
+ "rand",
+ "thiserror",
+ "tokio",
+ "winapi",
+]
+
 [[package]]
 name = "winapi"
 version = "0.3.9"

gst-meet/Cargo.toml

@@ -23,6 +23,7 @@ cocoa = { version = "0.24", default-features = false }
 
 [features]
 default = ["tls-rustls-native-roots"]
+log-rtp = ["lib-gst-meet/log-rtp"]
 tls-insecure = ["lib-gst-meet/tls-insecure"]
 tls-native = ["lib-gst-meet/tls-native"]
 tls-native-vendored = ["lib-gst-meet/tls-native-vendored"]

gst-meet/src/main.rs

@@ -58,6 +58,12 @@ struct Opt {
     help = "Disable TLS certificate verification (use with extreme caution)"
   )]
   tls_insecure: bool,
+  #[cfg(feature = "log-rtp")]
+  #[structopt(
+    long,
+    help = "Log all RTP/RTCP packets at DEBUG level"
+  )]
+  log_rtp: bool,
 }
 
 #[cfg(not(target_os = "macos"))]
@@ -154,6 +160,8 @@ async fn main_inner() -> Result<()> {
     region,
     video_codec,
     recv_pipeline_participant_template,
+    #[cfg(feature = "log-rtp")]
+    log_rtp,
     ..
   } = opt;
 
@@ -164,6 +172,10 @@ async fn main_inner() -> Result<()> {
     region,
     video_codec,
     extra_muc_features: vec![],
+    start_bitrate: 800,
+    stereo: false,
+    #[cfg(feature = "log-rtp")]
+    log_rtp,
   };
 
   let main_loop = glib::MainLoop::new(None, false);

lib-gst-meet-c/Cargo.toml

@@ -17,3 +17,7 @@ tracing = { version = "0.1", default-features = false }
 [lib]
 name = "gstmeet"
 crate_type = ["staticlib", "cdylib"]
+
+[features]
+default = []
+log-rtp = ["lib-gst-meet/log-rtp"]

lib-gst-meet-c/src/lib.rs

@@ -156,6 +156,13 @@ pub unsafe extern "C" fn gstmeet_connection_join_conference(
       .to_string_lossy()
       .to_string(),
     extra_muc_features: vec![],
+
+    // TODO
+    start_bitrate: 800,
+    stereo: false,
+    
+    #[cfg(feature = "log-rtp")]
+    log_rtp: false,
   };
   (*context)
     .runtime

lib-gst-meet/Cargo.toml

@@ -32,6 +32,7 @@ pem = { version = "1", default-features = false }
 rand = { version = "0.8", default-features = false, features = ["std", "std_rng"] }
 rcgen = { version = "0.9", default-features = false }
 ring = { version = "0.16", default-features = false }
+rtcp = { version = "0.6", default-features = false, optional = true }
 rustls = { version = "0.20", default-features = false, features = ["logging", "tls12"], optional = true }
 rustls-native-certs = { version = "0.6", default-features = false, optional = true }
 serde = { version = "1", default-features = false, features = ["derive"] }
@@ -55,6 +56,7 @@ xmpp-parsers = { git = "https://gitlab.com/xmpp-rs/xmpp-rs.git", default-feature
 # Ideally we would enable rustls/dangerous_configuration only when tls-insecure is enabled, but until weak-dep-features is stabilised, that
 # would cause rustls to always be pulled in.
 default = ["tls-rustls-native-roots"]
+log-rtp = ["rtcp"]
 tls-insecure = []
 tls-native = ["tokio-tungstenite/native-tls", "native-tls"]
 tls-native-vendored = ["tokio-tungstenite/native-tls-vendored", "native-tls/vendored"]

lib-gst-meet/src/conference.rs

@@ -75,6 +75,10 @@ pub struct JitsiConferenceConfig {
   pub region: Option<String>,
   pub video_codec: String,
   pub extra_muc_features: Vec<String>,
+  pub start_bitrate: u32,
+  pub stereo: bool,
+  #[cfg(feature = "log-rtp")]
+  pub log_rtp: bool,
 }
 
 #[derive(Clone)]
@@ -140,10 +144,8 @@ impl JitsiConference {
       machine_uid: Uuid::new_v4().to_string(),
       room: config.muc.to_string(),
       properties: hashmap! {
-        // Disable voice processing
+        "stereo".to_string() => config.stereo.to_string(),
-        // TODO put this in config
+        "startBitrate".to_string() => config.start_bitrate.to_string(),
-        "stereo".to_string() => "true".to_string(),
-        "startBitrate".to_string() => "800".to_string(),
       },
     };
 

lib-gst-meet/src/jingle.rs

@@ -5,6 +5,8 @@ use futures::stream::StreamExt;
 use glib::{ObjectExt, ToValue};
 use gstreamer::prelude::{ElementExt, GObjectExtManualGst, GstBinExt, PadExt};
 use gstreamer_rtp::{prelude::RTPHeaderExtensionExt, RTPHeaderExtension};
+#[cfg(feature = "log-rtp")]
+use gstreamer_rtp::RTPBuffer;
 use jitsi_xmpp_parsers::{
   jingle::{Action, Content, Description, Jingle, Transport},
   jingle_dtls_srtp::Fingerprint,
@@ -126,7 +128,6 @@ impl Codec {
 struct ParsedRtpDescription {
   codecs: Vec<Codec>,
   audio_hdrext_ssrc_audio_level: Option<u16>,
-  audio_hdrext_transport_cc: Option<u16>,
   video_hdrext_transport_cc: Option<u16>,
 }
 
@@ -190,7 +191,6 @@ impl JingleSession {
     let mut vp8 = None;
     let mut vp9 = None;
     let mut audio_hdrext_ssrc_audio_level = None;
-    let mut audio_hdrext_transport_cc = None;
     let mut video_hdrext_transport_cc = None;
 
     if description.media == "audio" {
@@ -209,9 +209,6 @@ impl JingleSession {
         if hdrext.uri == RTP_HDREXT_SSRC_AUDIO_LEVEL {
           audio_hdrext_ssrc_audio_level = Some(hdrext.id);
         }
-        else if hdrext.uri == RTP_HDREXT_TRANSPORT_CC {
-          audio_hdrext_transport_cc = Some(hdrext.id);
-        }
       }
     }
     else if description.media == "video" {
@@ -274,8 +271,6 @@ impl JingleSession {
         }
       }
       for hdrext in description.hdrexts.iter() {
-        // TODO: .parse::<u8>() won’t be needed after updating xmpp-parsers, it is now a u16 as
-        // defined in the XEP and related RFC.
         if hdrext.uri == RTP_HDREXT_TRANSPORT_CC {
           video_hdrext_transport_cc = Some(hdrext.id);
         }
@@ -325,7 +320,6 @@ impl JingleSession {
     Ok(Some(ParsedRtpDescription {
       codecs,
       audio_hdrext_ssrc_audio_level,
-      audio_hdrext_transport_cc,
       video_hdrext_transport_cc,
     }))
   }
@@ -467,7 +461,6 @@ impl JingleSession {
     let mut ice_transport = None;
     let mut codecs = vec![];
     let mut audio_hdrext_ssrc_audio_level = None;
-    let mut audio_hdrext_transport_cc = None;
     let mut video_hdrext_transport_cc = None;
 
     let mut remote_ssrc_map = HashMap::new();
@@ -480,8 +473,6 @@ impl JingleSession {
           codecs.extend(description.codecs);
           audio_hdrext_ssrc_audio_level =
             audio_hdrext_ssrc_audio_level.or(description.audio_hdrext_ssrc_audio_level);
-          audio_hdrext_transport_cc =
-            audio_hdrext_transport_cc.or(description.audio_hdrext_transport_cc);
           video_hdrext_transport_cc =
             video_hdrext_transport_cc.or(description.video_hdrext_transport_cc);
         }
@@ -592,9 +583,6 @@ impl JingleSession {
                 if let Some(hdrext) = audio_hdrext_ssrc_audio_level {
                   caps = caps.field(&format!("extmap-{}", hdrext), RTP_HDREXT_SSRC_AUDIO_LEVEL);
                 }
-                if let Some(hdrext) = audio_hdrext_transport_cc {
-                  caps = caps.field(&format!("extmap-{}", hdrext), &RTP_HDREXT_TRANSPORT_CC);
-                }
               }
               else {
                 // A video codec, as the only audio codec we support is Opus.
@@ -829,14 +817,6 @@ impl JingleSession {
                 let hdrext = RTPHeaderExtension::create_from_uri(&ext_uri)
                   .context("failed to create hdrext")?;
                 hdrext.set_id(ext_id);
-                // if ext_uri == RTP_HDREXT_SSRC_AUDIO_LEVEL {
-                // }
-                // else if ext_uri == RTP_HDREXT_TRANSPORT_CC {
-                //   // hdrext.set_property("n-streams", 2u32);
-                // }
-                // else {
-                //   bail!("unknown rtp hdrext: {}", ext_uri);
-                // };
                 Ok::<_, anyhow::Error>(hdrext)
               };
               match f() {
@@ -941,14 +921,6 @@ impl JingleSession {
           let hdrext =
             RTPHeaderExtension::create_from_uri(&ext_uri).context("failed to create hdrext")?;
           hdrext.set_id(ext_id);
-          // if ext_uri == RTP_HDREXT_SSRC_AUDIO_LEVEL {
-          // }
-          // else if ext_uri == RTP_HDREXT_TRANSPORT_CC {
-          //   // hdrext.set_property("n-streams", 2u32);
-          // }
-          // else {
-          //   bail!("unknown rtp hdrext: {}", ext_uri);
-          // }
           Ok::<_, anyhow::Error>(hdrext)
         };
         match f() {
@@ -995,12 +967,6 @@ impl JingleSession {
           let hdrext =
             RTPHeaderExtension::create_from_uri(&ext_uri).context("failed to create hdrext")?;
           hdrext.set_id(ext_id);
-          if ext_uri == RTP_HDREXT_TRANSPORT_CC {
-            // hdrext.set_property("n-streams", 2u32);
-          }
-          else {
-            bail!("unknown rtp hdrext: {}", ext_uri);
-          }
           Ok::<_, anyhow::Error>(hdrext)
         };
         match f() {
@@ -1029,13 +995,79 @@ impl JingleSession {
     debug!("linking rtpfunnel -> rtpbin");
     rtpfunnel.link_pads(None, &rtpbin, Some("send_rtp_sink_0"))?;
 
+    let rtp_recv_identity = gstreamer::ElementFactory::make("identity", None)?;
+    pipeline.add(&rtp_recv_identity)?;
+    let rtcp_recv_identity = gstreamer::ElementFactory::make("identity", None)?;
+    pipeline.add(&rtcp_recv_identity)?;
+    let rtp_send_identity = gstreamer::ElementFactory::make("identity", None)?;
+    pipeline.add(&rtp_send_identity)?;
+    let rtcp_send_identity = gstreamer::ElementFactory::make("identity", None)?;
+    pipeline.add(&rtcp_send_identity)?;
+
+    #[cfg(feature = "log-rtp")]
+    if conference.config.log_rtp {
+      let make_rtp_logger = |direction: &'static str| {
+        move |values: &[glib::Value]| -> Option<glib::Value> {
+          let f = || {
+            let buffer: gstreamer::Buffer = values[1].get()?;
+            let rtp_buffer = RTPBuffer::from_buffer_readable(&buffer)?;
+            debug!(
+              ssrc=rtp_buffer.ssrc(),
+              seq=rtp_buffer.seq(),
+              ts=rtp_buffer.timestamp(),
+              marker=rtp_buffer.is_marker(),
+              payload_size=rtp_buffer.payload_size(),
+              "RTP {}",
+              direction,
+            );
+            Ok::<_, anyhow::Error>(())
+          };
+          if let Err(e) = f() {
+            warn!("RTP {}: {:?}", direction, e);
+          }
+          None
+        }
+      };
+
+      let make_rtcp_logger = |direction: &'static str| {
+        move |values: &[glib::Value]| -> Option<glib::Value> {
+          let f = || {
+            let buffer: gstreamer::Buffer = values[1].get()?;
+            let mut buf = [0u8; 1500];
+            buffer.copy_to_slice(0, &mut buf[..buffer.size()]).map_err(|_| anyhow!("invalid RTCP packet size"))?;
+            let decoded = rtcp::packet::unmarshal(&mut &buf[..buffer.size()])?;
+            debug!(
+              "RTCP {} size={}\n{:?}",
+              direction,
+              buffer.size(),
+              decoded,
+            );
+            Ok::<_, anyhow::Error>(())
+          };
+          if let Err(e) = f() {
+            warn!("RTCP {}: {:?}", direction, e);
+          }
+          None
+        }
+      };
+
+      rtp_recv_identity.connect("handoff", false, make_rtp_logger("RECV"));
+      rtp_send_identity.connect("handoff", false, make_rtp_logger("SEND"));
+      rtcp_recv_identity.connect("handoff", false, make_rtcp_logger("RECV"));
+      rtcp_send_identity.connect("handoff", false, make_rtcp_logger("SEND"));
+    }
+
     debug!("link dtlssrtpdec -> rtpbin");
-    dtlssrtpdec.link_pads(Some("rtp_src"), &rtpbin, Some("recv_rtp_sink_0"))?;
+    dtlssrtpdec.link_pads(Some("rtp_src"), &rtp_recv_identity, None)?;
-    dtlssrtpdec.link_pads(Some("rtcp_src"), &rtpbin, Some("recv_rtcp_sink_0"))?;
+    rtp_recv_identity.link_pads(None, &rtpbin, Some("recv_rtp_sink_0"))?;
+    dtlssrtpdec.link_pads(Some("rtcp_src"), &rtcp_recv_identity, None)?;
+    rtcp_recv_identity.link_pads(None, &rtpbin, Some("recv_rtcp_sink_0"))?;
 
     debug!("linking rtpbin -> dtlssrtpenc");
-    rtpbin.link_pads(Some("send_rtp_src_0"), &dtlssrtpenc, Some("rtp_sink_0"))?;
+    rtpbin.link_pads(Some("send_rtp_src_0"), &rtp_send_identity, None)?;
-    rtpbin.link_pads(Some("send_rtcp_src_0"), &dtlssrtpenc, Some("rtcp_sink_0"))?;
+    rtp_send_identity.link_pads(None, &dtlssrtpenc, Some("rtp_sink_0"))?;
+    rtpbin.link_pads(Some("send_rtcp_src_0"), &rtcp_send_identity, None)?;
+    rtcp_send_identity.link_pads(None, &dtlssrtpenc, Some("rtcp_sink_0"))?;
 
     debug!("linking ice src -> dtlssrtpdec");
     nicesrc.link(&dtlssrtpdec)?;
@@ -1183,11 +1215,6 @@ impl JingleSession {
             RTP_HDREXT_SSRC_AUDIO_LEVEL.to_owned(),
           ));
         }
-        if let Some(hdrext) = audio_hdrext_transport_cc {
-          description
-            .hdrexts
-            .push(RtpHdrext::new(hdrext, RTP_HDREXT_TRANSPORT_CC.to_owned()));
-        }
       }
       else if initiate_content.name.0 == "video" {
         if let Some(hdrext) = video_hdrext_transport_cc {