support udp with retina or ffmpeg

CHANGELOG.md

@@ -13,8 +13,8 @@ Each release is tagged in Git and on the Docker repository
     before.
 *   fix [#147](https://github.com/scottlamb/moonfire-nvr/issues/147): confusing
     `nvr init` failures when using very old versions of SQLite.
-*   improve compatibility with Reolink cameras when using the default
-    `--rtsp-library=retina`.
+*   support `--rtsp-transport=udp`, which improves compatibility with Reolink
+    cameras.
 
 ## `v0.6.5` (2021-08-13)
 

server/Cargo.lock

@@ -1871,9 +1871,9 @@ dependencies = [
 
 [[package]]
 name = "retina"
-version = "0.2.0"
+version = "0.3.0"
 source = "registry+https://github.com/rust-lang/crates.io-index"
-checksum = "954bff9dec477d485c8fba6433f120b570e8fac582e2ea19d7c0c900a36c38e4"
+checksum = "446070f3caf291e982d240c7f921837f6da0cbffbc26f1d785b0a8d214f2cadd"
 dependencies = [
  "base64",
  "bitreader",
@@ -1886,6 +1886,7 @@ dependencies = [
  "once_cell",
  "pin-project",
  "pretty-hex",
+ "rand",
  "rtp-rs",
  "rtsp-types",
  "sdp",

server/Cargo.toml

@@ -46,7 +46,7 @@ nom = "6.0.0"
 parking_lot = { version = "0.11.1", features = [] }
 protobuf = { git = "https://github.com/stepancheg/rust-protobuf" }
 reffers = "0.6.0"
-retina = "0.2.0"
+retina = "0.3.0"
 ring = "0.16.2"
 rusqlite = "0.25.3"
 serde = { version = "1.0", features = ["derive"] }

server/src/cmds/config/cameras.rs

@@ -157,6 +157,7 @@ fn press_test_inner(
             url,
             username,
             password,
+            transport: retina::client::Transport::Tcp,
         },
     )?;
     Ok(format!(

server/src/cmds/run.rs

@@ -76,6 +76,9 @@ pub struct Args {
     /// developed by Moonfire NVR's author).
     #[structopt(long, default_value = "retina", parse(try_from_str))]
     rtsp_library: crate::stream::RtspLibrary,
+
+    #[structopt(long, default_value)]
+    rtsp_transport: retina::client::Transport,
 }
 
 // These are used in a hack to get the name of the current time zone (e.g. America/Los_Angeles).
@@ -223,6 +226,7 @@ async fn async_run(args: &Args) -> Result<i32, Error> {
         let env = streamer::Environment {
             db: &db,
             opener: args.rtsp_library.opener(),
+            transport: args.rtsp_transport,
             shutdown: &shutdown_streamers,
         };
 

server/src/stream.rs

@@ -9,7 +9,7 @@ use failure::{bail, Error};
 use futures::StreamExt;
 use lazy_static::lazy_static;
 use log::warn;
-use retina::client::Credentials;
+use retina::client::{Credentials, Transport};
 use retina::codec::{CodecItem, VideoParameters};
 use std::convert::TryFrom;
 use std::ffi::CString;
@@ -61,6 +61,7 @@ pub enum Source<'a> {
         url: Url,
         username: Option<String>,
         password: Option<String>,
+        transport: Transport,
     },
 }
 
@@ -71,6 +72,7 @@ pub enum Source {
         url: Url,
         username: Option<String>,
         password: Option<String>,
+        transport: Transport,
     },
 }
 
@@ -138,10 +140,17 @@ impl Opener for Ffmpeg {
                 url,
                 username,
                 password,
+                transport,
             } => {
                 let mut open_options = ffmpeg::avutil::Dictionary::new();
                 open_options
-                    .set(cstr!("rtsp_transport"), cstr!("tcp"))
+                    .set(
+                        cstr!("rtsp_transport"),
+                        match transport {
+                            Transport::Tcp => cstr!("tcp"),
+                            Transport::Udp => cstr!("udp"),
+                        },
+                    )
                     .unwrap();
                 open_options
                     .set(cstr!("user-agent"), cstr!("moonfire-nvr"))
@@ -284,28 +293,32 @@ impl Opener for RetinaOpener {
         let (startup_tx, startup_rx) = tokio::sync::oneshot::channel();
         let (frame_tx, frame_rx) = tokio::sync::mpsc::channel(1);
         let handle = tokio::runtime::Handle::current();
-        let (url, username, password) = match src {
+        let (url, options) = match src {
             #[cfg(test)]
             Source::File(_) => bail!("Retina doesn't support .mp4 files"),
             Source::Rtsp {
                 url,
                 username,
                 password,
-            } => (url, username, password),
-        };
-        let creds = match (username, password) {
-            (None, None) => None,
-            (Some(username), Some(password)) => Some(Credentials { username, password }),
-            (Some(username), None) => Some(Credentials {
-                username,
-                password: String::new(),
-            }),
-            _ => bail!("must supply username when supplying password"),
+                transport,
+            } => (
+                url,
+                retina::client::SessionOptions::default()
+                    .creds(match (username, password) {
+                        (None, None) => None,
+                        (Some(username), password) => Some(Credentials {
+                            username,
+                            password: password.unwrap_or_default(),
+                        }),
+                        _ => bail!("must supply username when supplying password"),
+                    })
+                    .transport(transport)
+                    .user_agent(format!("Moonfire NVR {}", env!("CARGO_PKG_VERSION"))),
+            ),
         };
 
-        // TODO: connection timeout.
         handle.spawn(async move {
-            let r = tokio::time::timeout(RETINA_TIMEOUT, RetinaOpener::play(url, creds)).await;
+            let r = tokio::time::timeout(RETINA_TIMEOUT, RetinaOpener::play(url, options)).await;
             let (mut session, video_params, first_frame) =
                 match r.unwrap_or_else(|_| Err(format_err!("timeout opening stream"))) {
                     Err(e) => {
@@ -378,7 +391,7 @@ impl RetinaOpener {
     /// Plays to first frame. No timeout; that's the caller's responsibility.
     async fn play(
         url: Url,
-        creds: Option<Credentials>,
+        options: retina::client::SessionOptions,
     ) -> Result<
         (
             Pin<Box<retina::client::Demuxed>>,
@@ -387,14 +400,7 @@ impl RetinaOpener {
         ),
         Error,
     > {
-        let mut session = retina::client::Session::describe(
-            url,
-            retina::client::SessionOptions::default()
-                .creds(creds)
-                .user_agent("Moonfire NVR".to_owned())
-                .ignore_spurious_data(true), // TODO: make this configurable.
-        )
-        .await?;
+        let mut session = retina::client::Session::describe(url, options).await?;
         let (video_i, mut video_params) = session
             .streams()
             .iter()

server/src/streamer.rs

@@ -20,6 +20,7 @@ where
     C: Clocks + Clone,
 {
     pub opener: &'a dyn stream::Opener,
+    pub transport: retina::client::Transport,
     pub db: &'tmp Arc<Database<C>>,
     pub shutdown: &'tmp Arc<AtomicBool>,
 }
@@ -39,6 +40,7 @@ where
     dir: Arc<dir::SampleFileDir>,
     syncer_channel: writer::SyncerChannel<::std::fs::File>,
     opener: &'a dyn stream::Opener,
+    transport: retina::client::Transport,
     stream_id: i32,
     short_name: String,
     url: Url,
@@ -72,6 +74,7 @@ where
             dir,
             syncer_channel,
             opener: env.opener,
+            transport: env.transport,
             stream_id,
             short_name: format!("{}-{}", c.short_name, s.type_.as_str()),
             url,
@@ -115,6 +118,7 @@ where
                     url: self.url.clone(),
                     username: self.username.clone(),
                     password: self.password.clone(),
+                    transport: self.transport,
                 },
             )?
         };
@@ -360,6 +364,7 @@ mod tests {
             opener: &opener,
             db: &db.db,
             shutdown: &opener.shutdown,
+            transport: retina::client::Transport::Tcp,
         };
         let mut stream;
         {