Support providing RTP extensions in local tracks (#336)

Author: lookback-hugotunius

rtp/CHANGELOG.md

@@ -3,6 +3,7 @@
 ## Unreleased
 
 * Increased minimum support rust version to `1.60.0`.
+* Adds a new generic header extensions type `rtp::extension::HeaderExtension` which allows abstracting over all known extensions as well as custom extensions. [#336](https://github.com/webrtc-rs/webrtc/pull/336) by [@k0nserv](https://github.com/k0nserv).
 
 ## v0.6.7
 

rtp/src/extension/mod.rs

@@ -1,4 +1,91 @@
+use std::borrow::Cow;
+use std::fmt;
+
+use util::{Marshal, MarshalSize};
+
 pub mod abs_send_time_extension;
 pub mod audio_level_extension;
 pub mod transport_cc_extension;
 pub mod video_orientation_extension;
+
+/// A generic RTP header extension.
+pub enum HeaderExtension {
+    AbsSendTime(abs_send_time_extension::AbsSendTimeExtension),
+    AudioLevel(audio_level_extension::AudioLevelExtension),
+    TransportCc(transport_cc_extension::TransportCcExtension),
+    VideoOrientation(video_orientation_extension::VideoOrientationExtension),
+
+    /// A custom extension
+    Custom {
+        uri: Cow<'static, str>,
+        extension: Box<dyn Marshal + Send + Sync + 'static>,
+    },
+}
+
+impl HeaderExtension {
+    pub fn uri(&self) -> Cow<'static, str> {
+        use HeaderExtension::*;
+
+        match self {
+            AbsSendTime(_) => "http://www.webrtc.org/experiments/rtp-hdrext/abs-send-time".into(),
+            AudioLevel(_) => "urn:ietf:params:rtp-hdrext:ssrc-audio-level".into(),
+            TransportCc(_) => {
+                "http://www.ietf.org/id/draft-holmer-rmcat-transport-wide-cc-extensions-01".into()
+            }
+            VideoOrientation(_) => "urn:3gpp:video-orientation".into(),
+            Custom { uri, .. } => uri.clone(),
+        }
+    }
+
+    pub fn is_same(&self, other: &Self) -> bool {
+        use HeaderExtension::*;
+        match (self, other) {
+            (AbsSendTime(_), AbsSendTime(_)) => true,
+            (AudioLevel(_), AudioLevel(_)) => true,
+            (TransportCc(_), TransportCc(_)) => true,
+            (VideoOrientation(_), VideoOrientation(_)) => true,
+            (Custom { uri, .. }, Custom { uri: other_uri, .. }) => uri == other_uri,
+            _ => false,
+        }
+    }
+}
+
+impl MarshalSize for HeaderExtension {
+    fn marshal_size(&self) -> usize {
+        use HeaderExtension::*;
+        match self {
+            AbsSendTime(ext) => ext.marshal_size(),
+            AudioLevel(ext) => ext.marshal_size(),
+            TransportCc(ext) => ext.marshal_size(),
+            VideoOrientation(ext) => ext.marshal_size(),
+            Custom { extension: ext, .. } => ext.marshal_size(),
+        }
+    }
+}
+
+impl Marshal for HeaderExtension {
+    fn marshal_to(&self, buf: &mut [u8]) -> util::Result<usize> {
+        use HeaderExtension::*;
+        match self {
+            AbsSendTime(ext) => ext.marshal_to(buf),
+            AudioLevel(ext) => ext.marshal_to(buf),
+            TransportCc(ext) => ext.marshal_to(buf),
+            VideoOrientation(ext) => ext.marshal_to(buf),
+            Custom { extension: ext, .. } => ext.marshal_to(buf),
+        }
+    }
+}
+
+impl fmt::Debug for HeaderExtension {
+    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
+        use HeaderExtension::*;
+
+        match self {
+            AbsSendTime(ext) => f.debug_tuple("AbsSendTime").field(ext).finish(),
+            AudioLevel(ext) => f.debug_tuple("AudioLevel").field(ext).finish(),
+            TransportCc(ext) => f.debug_tuple("TransportCc").field(ext).finish(),
+            VideoOrientation(ext) => f.debug_tuple("VideoOrientation").field(ext).finish(),
+            Custom { uri, extension: _ } => f.debug_struct("Custom").field("uri", uri).finish(),
+        }
+    }
+}

webrtc/CHANGELOG.md

@@ -1,4 +1,4 @@
-# webrtc-rc changelog
+# webrtc-rs changelog
 
 ## Unreleased
 
@@ -11,14 +11,16 @@ directions that should not send. [#316](https://github.com/webrtc-rs/webrtc/pull
 * Fixed a panic that would sometimes happen when collecting stats. [#327](https://github.com/webrtc-rs/webrtc/pull/327) by [@k0nserv](https://github.com/k0nserv).
 * Added new extension marshaller/unmarshaller for VideoOrientation, and made marshallers serializable via serde [#331](https://github.com/webrtc-rs/webrtc/pull/331) [#332](https://github.com/webrtc-rs/webrtc/pull/332)
 * Updated minimum rust version to `1.60.0`
-
+* Added a new `write_rtp_with_extensions` method to `TrackLocalStaticSample` and `TrackLocalStatiRTP`. [#336](https://github.com/webrtc-rs/webrtc/pull/336) by [@k0nserv](https://github.com/k0nserv).
+* Added a new `sample_writer` helper to `TrackLocalStaticSample`. [#336](https://github.com/webrtc-rs/webrtc/pull/336) by [@k0nserv](https://github.com/k0nserv).
 
 #### Breaking changes
 
 * Allow one single direction for extmap matching. [#321](https://github.com/webrtc-rs/webrtc/pull/321). API
 change for MediaEngine::register_header_extension
 * Removes support for Plan-B. All major implementations of WebRTC now support unified and continuing support for plan-b is an undue maintenance burden when unified can be used. See [“Unified Plan” Transition Guide (JavaScript)](https://docs.google.com/document/d/1-ZfikoUtoJa9k-GZG1daN0BU3IjIanQ_JSscHxQesvU/) for an overview of the changes required to migrate. [#320](https://github.com/webrtc-rs/webrtc/pull/320) by [@algesten](https://github.com/algesten).
 
+
 ## 0.5.1
 
 * Promote agent lock in ice_gather.rs create_agent() to top level of the function to avoid a race condition. [#290 Promote create_agent lock to top of function, to avoid race condition](https://github.com/webrtc-rs/webrtc/pull/290) contributed by [efer-ms](https://github.com/efer-ms)

webrtc/src/track/track_local/mod.rs

@@ -104,6 +104,7 @@ pub(crate) struct TrackBinding {
     id: String,
     ssrc: SSRC,
     payload_type: PayloadType,
+    params: RTCRtpParameters,
     write_stream: Option<Arc<dyn TrackLocalWriter + Send + Sync>>,
     sender_paused: Arc<AtomicBool>,
 }

webrtc/src/track/track_local/track_local_static_rtp.rs

@@ -1,7 +1,11 @@
+use std::collections::HashMap;
+
 use super::*;
 
 use crate::error::flatten_errs;
+use bytes::BytesMut;
 use tokio::sync::Mutex;
+use util::{Marshal, MarshalSize};
 
 /// TrackLocalStaticRTP  is a TrackLocal that has a pre-set codec and accepts RTP Packets.
 /// If you wish to send a media.Sample use TrackLocalStaticSample
@@ -42,6 +46,91 @@ impl TrackLocalStaticRTP {
             .iter()
             .all(|b| b.sender_paused.load(Ordering::SeqCst))
     }
+
+    /// write_rtp_with_extensions writes a RTP Packet to the TrackLocalStaticRTP
+    /// If one PeerConnection fails the packets will still be sent to
+    /// all PeerConnections. The error message will contain the ID of the failed
+    /// PeerConnections so you can remove them
+    ///
+    /// If the RTCRtpSender direction is such that no packets should be sent, any call to this
+    /// function are blocked internally. Care must be taken to not increase the sequence number
+    /// while the sender is paused. While the actual _sending_ is blocked, the receiver will
+    /// miss out when the sequence number "rolls over", which in turn will break SRTP.
+    ///
+    /// Extensions that are already configured on the packet are overwritten by extensions in
+    /// `extensions`.
+    pub async fn write_rtp_with_extensions(
+        &self,
+        p: &rtp::packet::Packet,
+        extensions: &[rtp::extension::HeaderExtension],
+    ) -> Result<usize> {
+        let mut n = 0;
+        let mut write_errs = vec![];
+        let mut pkt = p.clone();
+
+        let bindings = {
+            let bindings = self.bindings.lock().await;
+            bindings.clone()
+        };
+        // Prepare the extensions data
+        let extension_data: HashMap<_, _> = extensions
+            .iter()
+            .flat_map(|extension| {
+                let buf = {
+                    let mut buf = BytesMut::with_capacity(extension.marshal_size());
+                    buf.resize(extension.marshal_size(), 0);
+                    if let Err(err) = extension.marshal_to(&mut buf) {
+                        write_errs.push(Error::Util(err));
+                        return None;
+                    }
+
+                    buf.freeze()
+                };
+
+                Some((extension.uri(), buf))
+            })
+            .collect();
+
+        for b in bindings.into_iter() {
+            if b.is_sender_paused() {
+                // See caveat in function doc.
+                continue;
+            }
+            pkt.header.ssrc = b.ssrc;
+            pkt.header.payload_type = b.payload_type;
+
+            for (uri, data) in extension_data.iter() {
+                if let Some(id) = b
+                    .params
+                    .header_extensions
+                    .iter()
+                    .find(|ext| &ext.uri == uri)
+                    .map(|ext| ext.id)
+                {
+                    if let Err(err) = pkt.header.set_extension(id as u8, data.clone()) {
+                        write_errs.push(Error::Rtp(err));
+                        continue;
+                    }
+                }
+            }
+
+            if let Some(write_stream) = &b.write_stream {
+                match write_stream.write_rtp(&pkt).await {
+                    Ok(m) => {
+                        n += m;
+                    }
+                    Err(err) => {
+                        write_errs.push(err);
+                    }
+                }
+            } else {
+                write_errs.push(Error::new("track binding has none write_stream".to_owned()));
+            }
+        }
+
+        flatten_errs(write_errs)?;
+        Ok(n)
+    }
 }
 
 #[async_trait]
@@ -63,6 +152,7 @@ impl TrackLocal for TrackLocalStaticRTP {
                     ssrc: t.ssrc(),
                     payload_type: codec.payload_type,
                     write_stream: t.write_stream(),
+                    params: t.params.clone(),
                     id: t.id(),
                     sender_paused: t.paused.clone(),
                 }));
@@ -133,37 +223,7 @@ impl TrackLocalWriter for TrackLocalStaticRTP {
     /// while the sender is paused. While the actual _sending_ is blocked, the receiver will
     /// miss out when the sequence number "rolls over", which in turn will break SRTP.
     async fn write_rtp(&self, p: &rtp::packet::Packet) -> Result<usize> {
-        let mut n = 0;
-        let mut write_errs = vec![];
-        let mut pkt = p.clone();
-
-        let bindings = {
-            let bindings = self.bindings.lock().await;
-            bindings.clone()
-        };
-        for b in bindings {
-            if b.is_sender_paused() {
-                // See caveat in function doc.
-                continue;
-            }
-            pkt.header.ssrc = b.ssrc;
-            pkt.header.payload_type = b.payload_type;
-            if let Some(write_stream) = &b.write_stream {
-                match write_stream.write_rtp(&pkt).await {
-                    Ok(m) => {
-                        n += m;
-                    }
-                    Err(err) => {
-                        write_errs.push(err);
-                    }
-                }
-            } else {
-                write_errs.push(Error::new("track binding has none write_stream".to_owned()));
-            }
-        }
-
-        flatten_errs(write_errs)?;
-        Ok(n)
+        self.write_rtp_with_extensions(p, &[]).await
     }
 
     /// write writes a RTP Packet as a buffer to the TrackLocalStaticRTP