parse byte ranges

1 files changed, 74 insertions, 7 deletions

diff --git a/server/src/routes/stream.rs b/server/src/routes/stream.rs
index 094fc6a..9d90eea 100644
--- a/server/src/routes/stream.rs
+++ b/server/src/routes/stream.rs
@@ -5,17 +5,20 @@
 */
 use super::ui::error::MyError;
 use crate::library::Library;
+use anyhow::Result;
 use anyhow::{anyhow, Context};
 use jellyremuxer::RemuxerContext;
-use log::debug;
 use log::warn;
-use rocket::http::Header;
+use log::{debug, info};
+use rocket::http::{Header, Status};
+use rocket::request::{self, FromRequest};
 use rocket::response;
 use rocket::response::Responder;
 use rocket::Request;
 use rocket::Response;
-use rocket::{get, http::ContentType, response::stream::ReaderStream, State};
+use rocket::{get, http::ContentType, State};
 use std::ops::Deref;
+use std::ops::Range;
 use std::path::PathBuf;
 use tokio::io::{duplex, DuplexStream};
 use tokio_util::io::SyncIoBridge;
@@ -40,7 +43,9 @@ pub fn r_stream(
     tracks: String,
     remuxer: &State<RemuxerContext>,
     library: &State<Library>,
-) -> Result<(ContentType, ReaderStream![DuplexStream]), MyError> {
+    range: Option<RequestRange>,
+) -> Result<StreamResponse, MyError> {
+    info!("stream request (range={range:?})");
     let (a, b) = duplex(1024);
     let path = path.to_str().unwrap().to_string();
     let item = library
@@ -69,20 +74,82 @@ pub fn r_stream(
         }
     });
     debug!("starting stream");
-    Ok((ContentType::WEBM, ReaderStream::one(a)))
+    Ok(StreamResponse { stream: a, range })
 }
 
 pub struct StreamResponse {
     stream: DuplexStream,
+    range: Option<RequestRange>,
 }
 
 #[rocket::async_trait]
 impl<'r> Responder<'r, 'static> for StreamResponse {
     fn respond_to(self, _: &'r Request<'_>) -> response::Result<'static> {
-        Response::build()
-            .header(Header::new("accept-ranges", "bytes"))
+        let mut b = Response::build();
+        if let Some(range) = self.range {
+            b.status(Status::PartialContent);
+            b.header(Header::new("content-range", range.to_cr_hv()));
+        }
+        b.header(Header::new("accept-ranges", "bytes"))
             .header(ContentType::WEBM)
             .streamed_body(self.stream)
             .ok()
     }
 }
+
+#[derive(Debug)]
+pub struct RequestRange(Vec<Range<Option<usize>>>);
+
+impl RequestRange {
+    pub fn to_cr_hv(&self) -> String {
+        assert_eq!(self.0.len(), 1);
+        format!(
+            "bytes {}-{}/*",
+            self.0[0]
+                .start
+                .map(|e| format!("{e}"))
+                .unwrap_or(String::new()),
+            self.0[0]
+                .end
+                .map(|e| format!("{e}"))
+                .unwrap_or(String::new())
+        )
+    }
+    pub fn from_hv(s: &str) -> Result<Self> {
+        Ok(Self(
+            s.strip_prefix("bytes=")
+                .ok_or(anyhow!("prefix expected"))?
+                .split(",")
+                .map(|s| {
+                    let (l, r) = s
+                        .split_once("-")
+                        .ok_or(anyhow!("range delimeter missing"))?;
+                    let km = |s: &str| {
+                        if s.is_empty() {
+                            Ok::<_, anyhow::Error>(None)
+                        } else {
+                            Ok(Some(s.parse()?))
+                        }
+                    };
+                    Ok(km(l)?..km(r)?)
+                })
+                .into_iter()
+                .collect::<Result<Vec<_>>>()?,
+        ))
+    }
+}
+
+#[rocket::async_trait]
+impl<'r> FromRequest<'r> for RequestRange {
+    type Error = anyhow::Error;
+
+    async fn from_request(req: &'r Request<'_>) -> request::Outcome<Self, Self::Error> {
+        match req.headers().get("range").next() {
+            Some(v) => match Self::from_hv(v) {
+                Ok(v) => rocket::outcome::Outcome::Success(v),
+                Err(e) => rocket::outcome::Outcome::Failure((Status::BadRequest, e)),
+            },
+            None => rocket::outcome::Outcome::Forward(()),
+        }
+    }
+}