server: improve PartCapture error handling

src/server.rs

@@ -14,6 +14,12 @@ use warp::{
 };
 use crate::app::{App, Payload};
 
+fn bad_gateway() -> Box<dyn warp::Reply> {
+    let reply = warp::reply();
+    let reply = warp::reply::with_status(reply, StatusCode::BAD_GATEWAY);
+    Box::new(reply)
+}
+
 struct PartStream {
     app: App,
 }
@@ -88,7 +94,7 @@ struct PartCapture {
 
 impl PartCapture {
     // async fn start(app: App) -> PartCapture {
-    async fn start((mut rx, stream_headers): (Receiver<Payload>, Arc<HeaderMap>)) -> Result<Box<dyn warp::Reply>, Infallible> {
+    async fn start(mut rx: Receiver<Payload>, stream_headers: Arc<HeaderMap>) -> Result<Box<dyn warp::Reply>, Infallible> {
         loop {
             match rx.recv().await {
                 Ok(output) => match output.as_ref() {
@@ -103,15 +109,14 @@ impl PartCapture {
                     // skip content Bytes
                     _ => {},
                 }
-                Err(e) => return Ok(Box::new(
+                Err(e) => return Ok(bad_gateway()),
-                    warp::reply::with_status(warp::reply(), warp::http::StatusCode::BAD_GATEWAY)
-                )),
             }
         }
     }
 }
 
 impl Reply for PartCapture {
+    // TODO: count up to option<content-length>
     fn into_response(self) -> Response {
         let body_stream = stream::unfold(self.rx, |mut rx| async move {
             match rx.recv().await {
@@ -145,12 +150,13 @@ pub async fn run(app: App) {
     let capture = warp::path!("capture.jpg")
         .map(move || app.clone())
         .and_then(|app: App| async move {
-            app.subscribe()
+            match (app.subscribe(), app.headers()) {
-                .and_then(|rx| app.headers()
+                (Some(rx), Some(headers)) =>
-                          .map(|headers| (rx, headers)))
+                    PartCapture::start(rx, headers).await,
-                .ok_or_else(|| warp::reject::not_found())
+                _ =>
-        })
+                    Ok(bad_gateway()),
-        .and_then(PartCapture::start);
+            }
+        });
     let routes = warp::get()
         .and(stream)
         .or(capture);