experimental UDP support (fixes #30)

* support setting transport to UDP
* breaking change: use new PacketContext rather than RtspMessageContext
  in places where an RTP/RTCP packet is expected, as it can now be over
  UDP instead of via RTSP interleaved data
* send teardown, which is important with UDP sessions. (It also will
  help somewhat with Reolink TCP.) Along the way, make Session Unpin
  so that teardown can consume it without tripping over tokio::pin!().
* refactor to shrink the amount of data used by Session and how much
  gets memmoved around on the stack, instead of further growing it.

Because of missing RTCP RR and reorder buffer support, this is only
appropriate for using on a LAN. That's enough for me right now.

CHANGELOG.md

@@ -1,5 +1,7 @@
-## unreleased
+## `v0.3.0` (unreleased)
 
+*   [#30](https://github.com/scottlamb/retina/issues/30): experimental UDP
+    support.
 *   [#22](https://github.com/scottlamb/retina/issues/22): fix handling of
     44.1 kHz AAC audio.
 

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "retina"
-version = "0.2.0"
+version = "0.3.0"
 authors = ["Scott Lamb <[email protected]>"]
 license = "MIT/Apache-2.0"
 edition = "2018"
@@ -22,6 +22,7 @@ log = "0.4.8"
 once_cell = "1.7.2"
 pin-project = "1.0.7"
 pretty-hex = "0.2.1"
+rand = "0.8.3"
 rtp-rs = "0.6.0"
 rtsp-types = "0.0.2"
 sdp = "0.1.4"

README.md

@@ -14,7 +14,8 @@ Progress:
 *   [x] client support
     *   [x] digest authentication.
     *   [x] RTP over TCP via RTSP interleaved channels.
-    *   [ ] RTP over UDP.
+    *   [x] RTP over UDP (experimental).
+    *   *   [ ] re-order buffer. (Out-of-order packets are dropped now.)
     *   [x] RTSP/1.0.
     *   [ ] RTSP/2.0.
     *   [ ] SRTP.

benches/depacketize.rs

@@ -4,7 +4,7 @@
 use std::num::NonZeroU16;
 
 use criterion::{criterion_group, criterion_main, Criterion};
-use retina::client::{rtp::StrictSequenceChecker, Timeline};
+use retina::client::{rtp::InorderParser, Timeline};
 use retina::codec::{CodecItem, Depacketizer};
 use std::convert::TryFrom;
 use std::io::Write;
@@ -24,15 +24,15 @@ fn h264_aac<F: FnMut(CodecItem) -> ()>(mut f: F) {
         Timeline::new(Some(0), 90_000, None).unwrap(),
     ];
     let mut rtps = [
-        StrictSequenceChecker::new(None, Some(1)),
-        StrictSequenceChecker::new(None, Some(1)),
+        InorderParser::new(None, Some(1)),
+        InorderParser::new(None, Some(1)),
     ];
     let mut depacketizers = [
         Depacketizer::new("audio", "mpeg4-generic", 12_000, NonZeroU16::new(2), Some("profile-level-id=1;mode=AAC-hbr;sizelength=13;indexlength=3;indexdeltalength=3;config=1490")).unwrap(),
         Depacketizer::new("video", "h264", 90_000, None, Some("packetization-mode=1;profile-level-id=42C01E;sprop-parameter-sets=Z0LAHtkDxWhAAAADAEAAAAwDxYuS,aMuMsg==")).unwrap(),
     ];
     let conn_ctx = retina::ConnectionContext::dummy();
-    let msg_ctx = retina::RtspMessageContext::dummy();
+    let pkt_ctx = retina::PacketContext::dummy();
     while !remaining.is_empty() {
         assert!(remaining.len() > 4);
         assert_eq!(remaining[0], b'$');
@@ -50,9 +50,8 @@ fn h264_aac<F: FnMut(CodecItem) -> ()>(mut f: F) {
         let pkt = match rtps[stream_id].rtp(
             &retina::client::SessionOptions::default(),
             &conn_ctx,
-            &msg_ctx,
+            &pkt_ctx,
             &mut timelines[stream_id],
-            channel_id,
             stream_id,
             data,
         ) {

examples/client/metadata.rs

@@ -29,12 +29,11 @@ pub async fn run(opts: Opts) -> Result<(), Error> {
         .position(|s| matches!(s.parameters(), Some(retina::codec::Parameters::Message(..))))
         .ok_or_else(|| anyhow!("couldn't find onvif stream"))?;
     session.setup(onvif_stream_i).await?;
-    let session = session
+    let mut session = session
         .play(retina::client::PlayOptions::default().ignore_zero_seq(true))
         .await?
         .demuxed()?;
 
-    tokio::pin!(session);
     tokio::pin!(stop);
     loop {
         tokio::select! {
@@ -51,5 +50,6 @@ pub async fn run(opts: Opts) -> Result<(), Error> {
             },
         }
     }
+    session.teardown().await?;
     Ok(())
 }

examples/client/mp4.rs

@@ -20,8 +20,11 @@
 use anyhow::{anyhow, bail, Error};
 use bytes::{Buf, BufMut, BytesMut};
 use futures::StreamExt;
-use log::info;
-use retina::codec::{AudioParameters, CodecItem, VideoParameters};
+use log::{info, warn};
+use retina::{
+    client::Transport,
+    codec::{AudioParameters, CodecItem, VideoParameters},
+};
 
 use std::convert::TryFrom;
 use std::io::SeekFrom;
@@ -60,6 +63,12 @@ pub struct Opts {
     #[structopt(long, name = "secs")]
     duration: Option<u64>,
 
+    /// The transport to use: `tcp` or `udp` (experimental).
+    ///
+    /// Note: `--allow-loss` is strongly recommended with `udp`.
+    #[structopt(default_value, long)]
+    transport: retina::client::Transport,
+
     /// Path to `.mp4` file to write.
     #[structopt(parse(try_from_str))]
     out: PathBuf,
@@ -564,13 +573,18 @@ impl<W: AsyncWrite + AsyncSeek + Send + Unpin> Mp4Writer<W> {
 }
 
 pub async fn run(opts: Opts) -> Result<(), Error> {
+    if matches!(opts.transport, Transport::Udp) && !opts.allow_loss {
+        warn!("Using --transport=udp without strongly recommended --allow-loss!");
+    }
+
     let creds = super::creds(opts.src.username, opts.src.password);
     let stop_signal = tokio::signal::ctrl_c();
     let mut session = retina::client::Session::describe(
         opts.src.url,
         retina::client::SessionOptions::default()
             .creds(creds)
             .user_agent("Retina mp4 example".to_owned())
+            .transport(opts.transport)
             .ignore_spurious_data(opts.ignore_spurious_data),
     )
     .await?;
@@ -604,7 +618,7 @@ pub async fn run(opts: Opts) -> Result<(), Error> {
     if let Some((i, _)) = audio_stream {
         session.setup(i).await?;
     }
-    let session = session
+    let mut session = session
         .play(
             retina::client::PlayOptions::default()
                 .initial_timestamp(opts.initial_timestamp)
@@ -629,7 +643,6 @@ pub async fn run(opts: Opts) -> Result<(), Error> {
         }
         None => futures::future::Either::Right(futures::future::pending()),
     };
-    tokio::pin!(session);
     tokio::pin!(stop_signal);
     tokio::pin!(sleep);
     loop {
@@ -654,6 +667,7 @@ pub async fn run(opts: Opts) -> Result<(), Error> {
             },
         }
     }
+    session.teardown().await?;
     mp4.finish().await?;
     Ok(())
 }

fuzz/fuzz_targets/depacketize_h264.rs

@@ -13,7 +13,7 @@ fuzz_target!(|data: &[u8]| {
     let mut timestamp = retina::Timestamp::new(0, NonZeroU32::new(90_000).unwrap(), 0).unwrap();
     let mut sequence_number: u16 = 0;
     let conn_ctx = retina::ConnectionContext::dummy();
-    let msg_ctx = retina::RtspMessageContext::dummy();
+    let pkt_ctx = retina::PacketContext::dummy();
     while data.has_remaining() {
         let hdr = data.get_u8();
         let ts_change = (hdr & 0b001) != 0;
@@ -30,8 +30,7 @@ fuzz_target!(|data: &[u8]| {
             timestamp = timestamp.try_add(1).unwrap();
         }
         let pkt = retina::client::rtp::Packet {
-            ctx: msg_ctx,
-            channel_id: 0,
+            ctx: pkt_ctx,
             stream_id: 0,
             timestamp,
             ssrc: 0,