Tentative first pass at making simulcast egest possible

examples/examples/simulcast/simulcast.rs

@@ -160,7 +160,7 @@ async fn main() -> Result<()> {
                 if let Some(track) = track {
                     println!("Track has started");
 
-                    let rid = track.rid().to_owned();
+                    let rid = track.rid().map_or(String::from(""), String::from);
                     let output_track = if let Some(output_track) = output_tracks.get(&rid) {
                         Arc::clone(output_track)
                     } else {
@@ -199,7 +199,7 @@ async fn main() -> Result<()> {
 
                     tokio::spawn(async move {
                         // Read RTP packets being sent to webrtc-rs
-                        println!("enter track loop {}", track.rid());
+                        println!("enter track loop {:?}", track.rid());
                         while let Ok((rtp, _)) = track.read_rtp().await {
                             if let Err(err) = output_track.write_rtp(&rtp).await {
                                 if Error::ErrClosedPipe != err {
@@ -210,7 +210,7 @@ async fn main() -> Result<()> {
                                 }
                             }
                         }
-                        println!("exit track loop {}", track.rid());
+                        println!("exit track loop {:?}", track.rid());
                     });
                 }
                 Box::pin(async {})

interceptor/src/stream_info.rs

@@ -11,6 +11,7 @@ pub struct RTPHeaderExtension {
 #[derive(Default, Debug, Clone)]
 pub struct StreamInfo {
     pub id: String,
+    pub rid: Option<String>,
     pub attributes: Attributes,
     pub ssrc: u32,
     pub payload_type: u8,

webrtc/src/error.rs

@@ -197,6 +197,29 @@ pub enum Error {
     #[error("new track must be of the same kind as previous")]
     ErrRTPSenderNewTrackHasIncorrectKind,
 
+    #[error("Cannot call replace on a sender with multiple tracks")]
+    ErrRTPSenderCannotReplaceSimulcast,
+
+    #[error("Sender does not have track for RID")]
+    ErrRTPSenderNoTrackForRID,
+
+    #[error("Sender cannot add encoding due to RID collision")]
+    ErrRTPSenderRIDCollision,
+
+    #[error("Sender cannot add encoding as provided track does not match base track")]
+    ErrRTPSenderBaseEncodingMismatch,
+
+    /// ErrRTPSenderStopped indicates the sender was already stopped
+    #[error("Sender has already been stopped")]
+    ErrRTPSenderStopped,
+
+    /// ErrRTPSenderRidNil indicates that the track RID was empty
+    #[error("Sender cannot add encoding as rid is empty")]
+    ErrRTPSenderRidNil,
+
+    #[error("Sender cannot add encoding as there is no base track")]
+    ErrRTPSenderNoBaseEncoding,
+
     /// ErrUnbindFailed indicates that a TrackLocal was not able to be unbind
     #[error("failed to unbind TrackLocal from PeerConnection")]
     ErrUnbindFailed,

webrtc/src/peer_connection/mod.rs

@@ -472,7 +472,7 @@ impl RTCPeerConnection {
                             };
 
                             let sender = match t.sender().await {
-                                Some(s) => s.clone(),
+                                Some(s) => s,
                                 None => {
                                     log::warn!(
                                         "RtpSender missing for transeceiver with sending direction {} for mid {}",
@@ -495,7 +495,6 @@ impl RTCPeerConnection {
                             if stream_ids.is_empty() {
                                 return true;
                             }
-
                             // different stream id
                             if dmsid.split_whitespace().next() != Some(&stream_ids[0]) {
                                 return true;

webrtc/src/peer_connection/peer_connection_internal.rs

@@ -173,9 +173,9 @@ impl PeerConnectionInternal {
                     let mut receiver_needs_stopped = false;
 
                     for t in tracks {
-                        if !t.rid().is_empty() {
+                        if let Some(rid) = t.rid() {
                             if let Some(details) =
-                                track_details_for_rid(&track_details, t.rid().to_owned())
+                                track_details_for_rid(&track_details, String::from(rid))
                             {
                                 t.set_id(details.id.clone()).await;
                                 t.set_stream_id(details.stream_id.clone()).await;
@@ -1141,6 +1141,7 @@ impl PeerConnectionInternal {
                     let stream_info = create_stream_info(
                         "".to_owned(),
                         ssrc,
+                        None,
                         params.codecs[0].payload_type,
                         params.codecs[0].capability.clone(),
                         &params.header_extensions,
@@ -1182,7 +1183,7 @@ impl PeerConnectionInternal {
                                     return receiver
                                         .receive_for_rtx(
                                             0,
-                                            rsid,
+                                            &rsid,
                                             TrackStream {
                                                 stream_info: Some(stream_info.clone()),
                                                 rtp_read_stream,
@@ -1196,7 +1197,7 @@ impl PeerConnectionInternal {
 
                                 let track = receiver
                                     .receive_for_rid(
-                                        rid,
+                                        &rid,
                                         params,
                                         TrackStream {
                                             stream_info: Some(stream_info.clone()),
@@ -1535,20 +1536,25 @@ impl PeerConnectionInternal {
                 None => continue,
             };
 
-            let track_id = track.id().to_string();
             let kind = match track.kind() {
                 RTPCodecType::Unspecified => continue,
                 RTPCodecType::Audio => "audio",
                 RTPCodecType::Video => "video",
             };
 
-            track_infos.push(TrackInfo {
-                track_id,
-                ssrc: sender.ssrc,
-                mid: mid.clone(),
-                rid: None,
-                kind,
-            });
+            let encodings = sender.track_encodings.read().await;
+            for e in encodings.iter() {
+                let track_id = track.rid().map_or(track.id().to_string(), |rid| {
+                    format!("{}-{}", track.id(), rid)
+                });
+                track_infos.push(TrackInfo {
+                    track_id,
+                    ssrc: e.ssrc,
+                    mid: mid.clone(),
+                    rid: track.rid().map(String::from),
+                    kind,
+                });
+            }
         }
 
         let stream_stats = self
@@ -1610,7 +1616,7 @@ impl PeerConnectionInternal {
                     kind,
                     packets_sent,
                     mid,
-                    rid,
+                    rid: rid.map(|a| a.to_owned()),
                     header_bytes_sent,
                     bytes_sent,
                     nack_count,

webrtc/src/peer_connection/sdp/mod.rs

@@ -535,13 +535,36 @@ pub(crate) async fn add_transceiver_sdp(
     for mt in transceivers {
         if let Some(sender) = mt.sender().await {
             if let Some(track) = sender.track().await {
-                media = media.with_media_source(
-                    sender.ssrc,
-                    track.stream_id().to_owned(), /* cname */
-                    track.stream_id().to_owned(), /* streamLabel */
-                    track.id().to_owned(),
-                );
+                if mt.direction().has_send() {
+                    let send_parameters = sender.get_parameters().await;
+                    // Get the different encodings expressed first
+                    for encoding in &send_parameters.encodings {
+                        media = media.with_media_source(
+                            encoding.ssrc,
+                            track.stream_id().to_owned(), /* cname */
+                            track.stream_id().to_owned(), /* streamLabel */
+                            track.id().to_owned(),
+                        );
+                    }
+
+                    // Then tell the world about simulcast
+                    if send_parameters.encodings.len() > 1 {
+                        let mut send_rids: Vec<String> = vec![];
 
+                        for e in &send_parameters.encodings {
+                            if let Some(rid) = &e.rid {
+                                let mut s: String = rid.clone();
+                                send_rids.push(rid.clone());
+                                s.push_str(" send");
+                                media = media.with_value_attribute("rid".into(), s);
+                            }
+                        }
+                        // Simulcast)
+                        let mut s: String = "send ".to_owned();
+                        s.push_str(send_rids.join(";").as_ref());
+                        media = media.with_value_attribute("simulcast".into(), s);
+                    }
+                }
                 // Send msid based on the configured track if we haven't already
                 // sent on this sender. If we have sent we must keep the msid line consistent, this
                 // is handled below.
@@ -553,12 +576,10 @@ pub(crate) async fn add_transceiver_sdp(
                             track.id()
                         ));
                     }
-
                     sender.set_initial_track_id(track.id().to_string())?;
                     break;
                 }
             }
-
             if !is_plan_b {
                 if let Some(track_id) = sender.initial_track_id() {
                     // After we have include an msid attribute in an offer it must stay the same for

webrtc/src/rtp_transceiver/mod.rs

@@ -93,7 +93,7 @@ pub struct RTCRtpRtxParameters {
 /// <http://draft.ortc.org/#dom-rtcrtpcodingparameters>
 #[derive(Default, Debug, Clone, Serialize, Deserialize)]
 pub struct RTCRtpCodingParameters {
-    pub rid: String,
+    pub rid: Option<String>,
     pub ssrc: SSRC,
     pub payload_type: PayloadType,
     pub rtx: RTCRtpRtxParameters,
@@ -132,6 +132,7 @@ pub struct RTCRtpTransceiverInit {
 pub(crate) fn create_stream_info(
     id: String,
     ssrc: SSRC,
+    rid: Option<&str>,
     payload_type: PayloadType,
     codec: RTCRtpCodecCapability,
     webrtc_header_extensions: &[RTCRtpHeaderExtensionParameters],
@@ -154,6 +155,7 @@ pub(crate) fn create_stream_info(
 
     StreamInfo {
         id,
+        rid: rid.map(String::from),
         attributes: Attributes::new(),
         ssrc,
         payload_type,

webrtc/src/rtp_transceiver/rtp_receiver/mod.rs

@@ -198,7 +198,7 @@ impl RTPReceiverInternal {
 
         let tracks = self.tracks.read().await;
         for t in &*tracks {
-            if t.track.rid() == rid {
+            if Some(rid) == t.track.rid() {
                 if let Some(rtcp_interceptor) = &t.stream.rtcp_interceptor {
                     let a = Attributes::new();
 
@@ -529,6 +529,7 @@ impl RTCRtpReceiver {
                     let stream_info = create_stream_info(
                         "".to_owned(),
                         encoding.ssrc,
+                        None,
                         0,
                         codec.clone(),
                         &global_params.header_extensions,
@@ -586,6 +587,7 @@ impl RTCRtpReceiver {
                 let stream_info = create_stream_info(
                     "".to_owned(),
                     rtx_ssrc,
+                    None,
                     0,
                     codec.clone(),
                     &global_params.header_extensions,
@@ -597,7 +599,7 @@ impl RTCRtpReceiver {
 
                 self.receive_for_rtx(
                     rtx_ssrc,
-                    "".to_owned(),
+                    "",
                     TrackStream {
                         stream_info: Some(stream_info),
                         rtp_read_stream,
@@ -654,7 +656,7 @@ impl RTCRtpReceiver {
         let mut encodings = vec![RTCRtpDecodingParameters::default(); encoding_size];
         for (i, encoding) in encodings.iter_mut().enumerate() {
             if incoming.rids.len() > i {
-                encoding.rid = incoming.rids[i].clone();
+                encoding.rid = Some(incoming.rids[i].clone());
             }
             if incoming.ssrcs.len() > i {
                 encoding.ssrc = incoming.ssrcs[i];
@@ -743,13 +745,13 @@ impl RTCRtpReceiver {
     /// It populates all the internal state for the given RID
     pub(crate) async fn receive_for_rid(
         &self,
-        rid: String,
+        rid: &str,
         params: RTCRtpParameters,
         stream: TrackStream,
     ) -> Result<Arc<TrackRemote>> {
         let mut tracks = self.internal.tracks.write().await;
         for t in &mut *tracks {
-            if t.track.rid() == rid {
+            if Some(rid) == t.track.rid() {
                 t.track.set_kind(self.kind);
                 if let Some(codec) = params.codecs.first() {
                     t.track.set_codec(codec.clone()).await;
@@ -771,13 +773,13 @@ impl RTCRtpReceiver {
     pub(crate) async fn receive_for_rtx(
         &self,
         ssrc: SSRC,
-        rsid: String,
+        rsid: &str,
         repair_stream: TrackStream,
     ) -> Result<()> {
         let mut tracks = self.internal.tracks.write().await;
         let l = tracks.len();
         for t in &mut *tracks {
-            if (ssrc != 0 && l == 1) || t.track.rid() == rsid {
+            if (ssrc != 0 && l == 1) || Some(rsid) == t.track.rid() {
                 t.repair_stream = repair_stream;
 
                 let receive_mtu = self.receive_mtu;