handle stale RTSP sessions

* upgrade to Retina 0.3.1 which automatically tears down sessions
* wait out stale sessions before reconnecting
* wait for teardown to complete before shutting down

This adds some pressure on #117: it will keep waiting for the stale
session to expire even if the user has requested shutdown. I'll try
to address that next.

server/src/cmds/config/cameras.rs

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

server/src/cmds/run.rs

@@ -10,6 +10,7 @@ use failure::{bail, Error, ResultExt};
 use fnv::FnvHashMap;
 use futures::future::FutureExt;
 use hyper::service::{make_service_fn, service_fn};
+use log::error;
 use log::{info, warn};
 use std::path::PathBuf;
 use std::sync::atomic::{AtomicBool, Ordering};
@@ -216,6 +217,8 @@ async fn async_run(args: &Args) -> Result<i32, Error> {
     // Start a streamer for each stream.
     let shutdown_streamers = Arc::new(AtomicBool::new(false));
     let mut streamers = Vec::new();
+    let mut session_groups_by_camera: FnvHashMap<i32, Arc<retina::client::SessionGroup>> =
+        FnvHashMap::default();
     let syncers = if !args.read_only {
         let l = db.lock();
         let mut dirs = FnvHashMap::with_capacity_and_hasher(
@@ -271,6 +274,10 @@ async fn async_run(args: &Args) -> Result<i32, Error> {
             };
             let rotate_offset_sec = streamer::ROTATE_INTERVAL_SEC * i as i64 / streams as i64;
             let syncer = syncers.get(&sample_file_dir_id).unwrap();
+            let session_group = session_groups_by_camera
+                .entry(camera.id)
+                .or_default()
+                .clone();
             let mut streamer = streamer::Streamer::new(
                 &env,
                 syncer.dir.clone(),
@@ -278,6 +285,7 @@ async fn async_run(args: &Args) -> Result<i32, Error> {
                 *id,
                 camera,
                 stream,
+                session_group,
                 rotate_offset_sec,
                 streamer::ROTATE_INTERVAL_SEC,
             )?;
@@ -344,6 +352,14 @@ async fn async_run(args: &Args) -> Result<i32, Error> {
 
     info!("Waiting for HTTP requests to finish.");
     server_handle.await??;
+
+    info!("Waiting for TEARDOWN requests to complete.");
+    for g in session_groups_by_camera.values() {
+        if let Err(e) = g.await_teardown().await {
+            error!("{}", e);
+        }
+    }
+
     info!("Exiting.");
     Ok(0)
 }

server/src/stream.rs

@@ -15,6 +15,7 @@ use std::convert::TryFrom;
 use std::ffi::CString;
 use std::pin::Pin;
 use std::result::Result;
+use std::sync::Arc;
 use url::Url;
 
 static START_FFMPEG: parking_lot::Once = parking_lot::Once::new();
@@ -62,6 +63,7 @@ pub enum Source<'a> {
         username: Option<String>,
         password: Option<String>,
         transport: Transport,
+        session_group: Arc<retina::client::SessionGroup>,
     },
 }
 
@@ -73,6 +75,7 @@ pub enum Source {
         username: Option<String>,
         password: Option<String>,
         transport: Transport,
+        session_group: Arc<retina::client::SessionGroup>,
     },
 }
 
@@ -141,6 +144,7 @@ impl Opener for Ffmpeg {
                 username,
                 password,
                 transport,
+                ..
             } => {
                 let mut open_options = ffmpeg::avutil::Dictionary::new();
                 open_options
@@ -301,6 +305,7 @@ impl Opener for RetinaOpener {
                 username,
                 password,
                 transport,
+                session_group,
             } => (
                 url,
                 retina::client::SessionOptions::default()
@@ -313,6 +318,7 @@ impl Opener for RetinaOpener {
                         _ => bail!("must supply username when supplying password"),
                     })
                     .transport(transport)
+                    .session_group(session_group)
                     .user_agent(format!("Moonfire NVR {}", env!("CARGO_PKG_VERSION"))),
             ),
         };

server/src/streamer.rs

@@ -42,6 +42,7 @@ where
     opener: &'a dyn stream::Opener,
     transport: retina::client::Transport,
     stream_id: i32,
+    session_group: Arc<retina::client::SessionGroup>,
     short_name: String,
     url: Url,
     username: Option<String>,
@@ -59,6 +60,7 @@ where
         stream_id: i32,
         c: &Camera,
         s: &Stream,
+        session_group: Arc<retina::client::SessionGroup>,
         rotate_offset_sec: i64,
         rotate_interval_sec: i64,
     ) -> Result<Self, Error> {
@@ -76,6 +78,7 @@ where
             opener: env.opener,
             transport: env.transport,
             stream_id,
+            session_group,
             short_name: format!("{}-{}", c.short_name, s.type_.as_str()),
             url,
             username: c.username.clone(),
@@ -110,6 +113,28 @@ where
         info!("{}: Opening input: {}", self.short_name, self.url.as_str());
         let clocks = self.db.clocks();
 
+        let mut waited = false;
+        loop {
+            let status = self.session_group.stale_sessions();
+            if let Some(max_expires) = status.max_expires {
+                if let Some(d) = max_expires.checked_duration_since(tokio::time::Instant::now()) {
+                    log::info!(
+                        "{}: Waiting {:?} for {} stale sessions to expire",
+                        &self.short_name,
+                        d,
+                        status.num_sessions
+                    );
+                    std::thread::sleep(d);
+                    waited = true;
+                }
+            } else {
+                if waited {
+                    log::info!("{}: Done waiting", &self.short_name);
+                }
+                break;
+            }
+        }
+
         let (extra_data, mut stream) = {
             let _t = TimerGuard::new(&clocks, || format!("opening {}", self.url.as_str()));
             self.opener.open(
@@ -119,6 +144,7 @@ where
                     username: self.username.clone(),
                     password: self.password.clone(),
                     transport: self.transport,
+                    session_group: self.session_group.clone(),
                 },
             )?
         };
@@ -383,6 +409,7 @@ mod tests {
                 testutil::TEST_STREAM_ID,
                 camera,
                 s,
+                Arc::new(retina::client::SessionGroup::default()),
                 0,
                 3,
             )