Outbound sequence number overriding (#378)

Author: logist322

webrtc/src/error.rs

@@ -197,6 +197,14 @@ pub enum Error {
     #[error("new track must be of the same kind as previous")]
     ErrRTPSenderNewTrackHasIncorrectKind,
 
+    /// ErrRTPSenderDataSent indicates that the sequence number transformer tries to be enabled after the data sending began
+    #[error("Sequence number transformer must be enabled before sending data")]
+    ErrRTPSenderDataSent,
+
+    /// ErrRTPSenderSeqTransEnabled indicates that the sequence number transformer has been already enabled
+    #[error("Sequence number transformer has been already enabled")]
+    ErrRTPSenderSeqTransEnabled,
+
     /// ErrUnbindFailed indicates that a TrackLocal was not able to be unbind
     #[error("failed to unbind TrackLocal from PeerConnection")]
     ErrUnbindFailed,

webrtc/src/rtp_transceiver/rtp_sender/mod.rs

@@ -22,6 +22,8 @@ use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Weak};
 use tokio::sync::{mpsc, Mutex, Notify};
 
+use super::srtp_writer_future::SequenceTransformer;
+
 pub(crate) struct RTPSenderInternal {
     pub(crate) send_called_rx: Mutex<mpsc::Receiver<()>>,
     pub(crate) stop_called_rx: Arc<Notify>,
@@ -81,6 +83,7 @@ pub struct RTCRtpSender {
 
     pub(crate) srtp_stream: Arc<SrtpWriterFuture>,
     pub(crate) stream_info: Mutex<StreamInfo>,
+    seq_trans: Arc<SequenceTransformer>,
 
     pub(crate) context: Mutex<TrackLocalContext>,
 
@@ -132,7 +135,7 @@ impl RTCRtpSender {
         media_engine: Arc<MediaEngine>,
         interceptor: Arc<dyn Interceptor + Send + Sync>,
         start_paused: bool,
-    ) -> RTCRtpSender {
+    ) -> Self {
         let id = generate_crypto_random_string(
             32,
             b"abcdefghijklmnopqrstuvwxyzABCDEFGHIJKLMNOPQRSTUVWXYZ",
@@ -150,13 +153,15 @@ impl RTCRtpSender {
             rtcp_interceptor: Mutex::new(None),
         });
 
+        let seq_trans = Arc::new(SequenceTransformer::new());
         let srtp_stream = Arc::new(SrtpWriterFuture {
             closed: AtomicBool::new(false),
             ssrc,
             rtp_sender: Arc::downgrade(&internal),
             rtp_transport: Arc::clone(&transport),
             rtcp_read_stream: Mutex::new(None),
             rtp_write_session: Mutex::new(None),
+            seq_trans: Arc::clone(&seq_trans),
         });
 
         let srtp_rtcp_reader = Arc::clone(&srtp_stream) as Arc<dyn RTCPReader + Send + Sync>;
@@ -170,11 +175,12 @@ impl RTCRtpSender {
             .as_ref()
             .map(|track| vec![track.stream_id().to_string()])
             .unwrap_or_default();
-        RTCRtpSender {
+        Self {
             track: Mutex::new(track),
 
             srtp_stream,
             stream_info: Mutex::new(StreamInfo::default()),
+            seq_trans,
 
             context: Mutex::new(TrackLocalContext::default()),
             transport,
@@ -327,6 +333,8 @@ impl RTCRtpSender {
         };
 
         let result = if let Some(t) = &track {
+            self.seq_trans.reset_offset();
+
             let new_context = TrackLocalContext {
                 id: context.id.clone(),
                 params: self
@@ -479,6 +487,18 @@ impl RTCRtpSender {
         self.internal.read_rtcp(self.receive_mtu).await
     }
 
+    /// Enables overriding outgoing `RTP` packets' `sequence number`s.
+    ///
+    /// Must be called once before any data sent or never called at all.
+    ///
+    /// # Errors
+    ///
+    /// Errors if this [`RTCRtpSender`] has started to send data or sequence
+    /// transforming has been already enabled.
+    pub fn enable_seq_transformer(&self) -> Result<()> {
+        self.seq_trans.enable()
+    }
+
     /// has_sent tells if data has been ever sent for this instance
     pub(crate) async fn has_sent(&self) -> bool {
         let send_called_tx = self.send_called_tx.lock().await;

webrtc/src/rtp_transceiver/srtp_writer_future.rs

@@ -12,6 +12,92 @@ use interceptor::{Attributes, RTCPReader, RTPWriter};
 use std::sync::atomic::{AtomicBool, Ordering};
 use std::sync::{Arc, Weak};
 use tokio::sync::Mutex;
+use util;
+
+/// `RTP` packet sequence number manager.
+///
+/// Used to override outgoing `RTP` packets' sequence numbers. On creating it is
+/// unabled and can be enabled before sending data begining. Once data sending
+/// began it can not be enabled any more.
+pub(crate) struct SequenceTransformer(util::sync::Mutex<SequenceTransformerInner>);
+
+/// [`SequenceTransformer`] inner.
+struct SequenceTransformerInner {
+    offset: u16,
+    last_sq: u16,
+    reset_needed: bool,
+    enabled: bool,
+    data_sent: bool,
+}
+
+impl SequenceTransformer {
+    /// Creates a new [`SequenceTransformer`].
+    pub(crate) fn new() -> Self {
+        Self(util::sync::Mutex::new(SequenceTransformerInner {
+            offset: 0,
+            last_sq: rand::random(),
+            reset_needed: false,
+            enabled: false,
+            data_sent: false,
+        }))
+    }
+
+    /// Enables this [`SequenceTransformer`].
+    ///
+    /// # Errors
+    ///
+    /// With [`Error::ErrRTPSenderSeqTransEnabled`] on trying to enable already
+    /// enabled [`SequenceTransformer`].
+    ///
+    /// With [`Error::ErrRTPSenderSeqTransEnabled`] on trying to enable
+    /// [`SequenceTransformer`] after data sending began.
+    pub(crate) fn enable(&self) -> Result<()> {
+        let mut guard = self.0.lock();
+
+        if guard.enabled {
+            return Err(Error::ErrRTPSenderSeqTransEnabled);
+        }
+
+        (!guard.data_sent)
+            .then(|| {
+                guard.enabled = true;
+            })
+            .ok_or(Error::ErrRTPSenderDataSent)
+    }
+
+    /// Indicates [`SequenceTransformer`] about necessity of recalculating
+    /// `offset`.
+    pub(crate) fn reset_offset(&self) {
+        self.0.lock().reset_needed = true;
+    }
+
+    /// Gets [`Some`] consistent `sequence number` if this [`SequenceTransformer`] is
+    /// enabled or [`None`] if it is not.
+    ///
+    /// Once this method is called, considers data sending began.
+    fn seq_number(&self, raw_sn: u16) -> Option<u16> {
+        let mut guard = self.0.lock();
+        guard.data_sent = true;
+
+        if !guard.enabled {
+            return None;
+        }
+
+        let offset = guard
+            .reset_needed
+            .then(|| {
+                guard.reset_needed = false;
+                let offset = guard.last_sq.overflowing_sub(raw_sn.overflowing_sub(1).0).0;
+                guard.offset = offset;
+                offset
+            })
+            .unwrap_or(guard.offset);
+        let next = raw_sn.overflowing_add(offset).0;
+        guard.last_sq = next;
+
+        Some(next)
+    }
+}
 
 /// SrtpWriterFuture blocks Read/Write calls until
 /// the SRTP Session is available
@@ -22,6 +108,7 @@ pub(crate) struct SrtpWriterFuture {
     pub(crate) rtp_transport: Arc<RTCDtlsTransport>,
     pub(crate) rtcp_read_stream: Mutex<Option<Arc<Stream>>>, // atomic.Value // *
     pub(crate) rtp_write_session: Mutex<Option<Arc<Session>>>, // atomic.Value // *
+    pub(crate) seq_trans: Arc<SequenceTransformer>,
 }
 
 impl SrtpWriterFuture {
@@ -181,6 +268,15 @@ impl RTCPReader for SrtpWriterFuture {
 #[async_trait]
 impl RTPWriter for SrtpWriterFuture {
     async fn write(&self, pkt: &rtp::packet::Packet, _a: &Attributes) -> IResult<usize> {
-        Ok(self.write_rtp(pkt).await?)
+        Ok(
+            match self.seq_trans.seq_number(pkt.header.sequence_number) {
+                Some(seq_num) => {
+                    let mut new_pkt = pkt.clone();
+                    new_pkt.header.sequence_number = seq_num;
+                    self.write_rtp(&new_pkt).await?
+                }
+                None => self.write_rtp(pkt).await?,
+            },
+        )
     }
 }