From ec4c1e0a441a6c475632de5244db9b889680471d Mon Sep 17 00:00:00 2001 From: Evan Almloff Date: Mon, 18 Dec 2023 08:32:50 -0600 Subject: [PATCH 1/2] add video stream example --- Cargo.toml | 1 + examples/video_stream.rs | 178 +++++++++++++++++++++++++++++++++++++++ 2 files changed, 179 insertions(+) create mode 100644 examples/video_stream.rs diff --git a/Cargo.toml b/Cargo.toml index a78f60d3e..41cd7cd83 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -133,3 +133,4 @@ fern = { version = "0.6.0", features = ["colored"] } env_logger = "0.10.0" simple_logger = "4.0.0" thiserror = { workspace = true } +http-range = "0.1.5" diff --git a/examples/video_stream.rs b/examples/video_stream.rs new file mode 100644 index 000000000..6338a7d68 --- /dev/null +++ b/examples/video_stream.rs @@ -0,0 +1,178 @@ +use dioxus::prelude::*; +use dioxus_desktop::wry::http; +use dioxus_desktop::wry::http::Response; +use dioxus_desktop::{use_asset_handler, AssetRequest}; +use http::{header::*, response::Builder as ResponseBuilder, status::StatusCode}; +use std::borrow::Cow; +use std::sync::{Arc, Mutex}; +use std::{ + io::{Read, Seek, SeekFrom, Write}, + path::PathBuf, +}; + +fn main() { + dioxus_desktop::launch(app); +} + +fn app(cx: Scope) -> Element { + let id = Arc::new(Mutex::new(0)); + use_asset_handler(cx, move |request: &AssetRequest| { + let response: Option>> = match get_stream_response(request, &id) + { + Ok(response) => Some(response.map(Cow::Owned)), + Err(err) => { + eprintln!("Error: {}", err); + None + } + }; + async move { response } + }); + + cx.render(rsx! { + div { + video { + src: "test_video.mp4", + autoplay: true, + controls: true, + width: 640, + height: 480, + } + } + }) +} + +fn get_stream_response( + request: &AssetRequest, + boundary_id: &Arc>, +) -> Result>, Box> { + let video_file = PathBuf::from("./examples/assets/test_video.mp4"); + + let mut file = std::fs::File::open(&video_file)?; + + // get file length + let len = { + let old_pos = file.stream_position()?; + let len = file.seek(SeekFrom::End(0))?; + file.seek(SeekFrom::Start(old_pos))?; + len + }; + + let mut resp = ResponseBuilder::new().header(CONTENT_TYPE, "video/mp4"); + + // if the webview sent a range header, we need to send a 206 in return + // Actually only macOS and Windows are supported. Linux will ALWAYS return empty headers. + let http_response = if let Some(range_header) = request.headers().get("range") { + let not_satisfiable = || { + ResponseBuilder::new() + .status(StatusCode::RANGE_NOT_SATISFIABLE) + .header(CONTENT_RANGE, format!("bytes */{len}")) + .body(vec![]) + }; + + // parse range header + let ranges = if let Ok(ranges) = http_range::HttpRange::parse(range_header.to_str()?, len) { + ranges + .iter() + // map the output back to spec range , example: 0-499 + .map(|r| (r.start, r.start + r.length - 1)) + .collect::>() + } else { + return Ok(not_satisfiable()?); + }; + + /// The Maximum bytes we send in one range + const MAX_LEN: u64 = 1000 * 1024; + + if ranges.len() == 1 { + let &(start, mut end) = ranges.first().unwrap(); + + // check if a range is not satisfiable + // + // this should be already taken care of by HttpRange::parse + // but checking here again for extra assurance + if start >= len || end >= len || end < start { + return Ok(not_satisfiable()?); + } + + // adjust end byte for MAX_LEN + end = start + (end - start).min(len - start).min(MAX_LEN - 1); + + // calculate number of bytes needed to be read + let bytes_to_read = end + 1 - start; + + // allocate a buf with a suitable capacity + let mut buf = Vec::with_capacity(bytes_to_read as usize); + // seek the file to the starting byte + file.seek(SeekFrom::Start(start))?; + // read the needed bytes + file.take(bytes_to_read).read_to_end(&mut buf)?; + + resp = resp.header(CONTENT_RANGE, format!("bytes {start}-{end}/{len}")); + resp = resp.header(CONTENT_LENGTH, end + 1 - start); + resp = resp.status(StatusCode::PARTIAL_CONTENT); + resp.body(buf) + } else { + let mut buf = Vec::new(); + let ranges = ranges + .iter() + .filter_map(|&(start, mut end)| { + // filter out unsatisfiable ranges + // + // this should be already taken care of by HttpRange::parse + // but checking here again for extra assurance + if start >= len || end >= len || end < start { + None + } else { + // adjust end byte for MAX_LEN + end = start + (end - start).min(len - start).min(MAX_LEN - 1); + Some((start, end)) + } + }) + .collect::>(); + + let mut id = boundary_id.lock().unwrap(); + *id += 1; + let boundary = format!("sadasq2e{id}"); + let boundary_sep = format!("\r\n--{boundary}\r\n"); + let boundary_closer = format!("\r\n--{boundary}\r\n"); + + resp = resp.header( + CONTENT_TYPE, + format!("multipart/byteranges; boundary={boundary}"), + ); + + for (end, start) in ranges { + // a new range is being written, write the range boundary + buf.write_all(boundary_sep.as_bytes())?; + + // write the needed headers `Content-Type` and `Content-Range` + buf.write_all(format!("{CONTENT_TYPE}: video/mp4\r\n").as_bytes())?; + buf.write_all( + format!("{CONTENT_RANGE}: bytes {start}-{end}/{len}\r\n").as_bytes(), + )?; + + // write the separator to indicate the start of the range body + buf.write_all("\r\n".as_bytes())?; + + // calculate number of bytes needed to be read + let bytes_to_read = end + 1 - start; + + let mut local_buf = vec![0_u8; bytes_to_read as usize]; + file.seek(SeekFrom::Start(start))?; + file.read_exact(&mut local_buf)?; + buf.extend_from_slice(&local_buf); + } + // all ranges have been written, write the closing boundary + buf.write_all(boundary_closer.as_bytes())?; + + resp.body(buf) + } + } else { + resp = resp.header(CONTENT_LENGTH, len); + let mut buf = Vec::with_capacity(len as usize); + file.read_to_end(&mut buf)?; + resp.body(buf) + }; + + http_response.map_err(Into::into) +} From dfb118e06ccc97885979da3bd8924de330b2078d Mon Sep 17 00:00:00 2001 From: Evan Almloff Date: Mon, 18 Dec 2023 09:34:30 -0600 Subject: [PATCH 2/2] make the video streaming async --- .gitignore | 1 + examples/video_stream.rs | 108 ++++++++++++++++--------------- packages/desktop/src/protocol.rs | 16 ++--- 3 files changed, 65 insertions(+), 60 deletions(-) diff --git a/.gitignore b/.gitignore index b8025753e..364e8d0fb 100644 --- a/.gitignore +++ b/.gitignore @@ -4,6 +4,7 @@ /dist Cargo.lock .DS_Store +/examples/assets/test_video.mp4 .vscode/* !.vscode/settings.json diff --git a/examples/video_stream.rs b/examples/video_stream.rs index 6338a7d68..f495299f4 100644 --- a/examples/video_stream.rs +++ b/examples/video_stream.rs @@ -4,56 +4,64 @@ use dioxus_desktop::wry::http::Response; use dioxus_desktop::{use_asset_handler, AssetRequest}; use http::{header::*, response::Builder as ResponseBuilder, status::StatusCode}; use std::borrow::Cow; -use std::sync::{Arc, Mutex}; -use std::{ - io::{Read, Seek, SeekFrom, Write}, - path::PathBuf, -}; +use std::{io::SeekFrom, path::PathBuf}; +use tokio::io::AsyncReadExt; +use tokio::io::AsyncSeekExt; +use tokio::io::AsyncWriteExt; + +const VIDEO_PATH: &str = "./examples/assets/test_video.mp4"; fn main() { + let video_file = PathBuf::from(VIDEO_PATH); + if !video_file.exists() { + tokio::runtime::Runtime::new() + .unwrap() + .block_on(async move { + println!("Downloading video file..."); + let video_url = + "http://commondatastorage.googleapis.com/gtv-videos-bucket/sample/BigBuckBunny.mp4"; + let mut response = reqwest::get(video_url).await.unwrap(); + let mut file = tokio::fs::File::create(&video_file).await.unwrap(); + while let Some(chunk) = response.chunk().await.unwrap() { + file.write_all(&chunk).await.unwrap(); + } + }); + } dioxus_desktop::launch(app); } fn app(cx: Scope) -> Element { - let id = Arc::new(Mutex::new(0)); use_asset_handler(cx, move |request: &AssetRequest| { - let response: Option>> = match get_stream_response(request, &id) - { - Ok(response) => Some(response.map(Cow::Owned)), - Err(err) => { - eprintln!("Error: {}", err); - None - } - }; - async move { response } + let request = request.clone(); + async move { + let video_file = PathBuf::from(VIDEO_PATH); + let mut file = tokio::fs::File::open(&video_file).await.unwrap(); + let response: Option>> = + match get_stream_response(&mut file, &request).await { + Ok(response) => Some(response.map(Cow::Owned)), + Err(err) => { + eprintln!("Error: {}", err); + None + } + }; + response + } }); - cx.render(rsx! { - div { - video { - src: "test_video.mp4", - autoplay: true, - controls: true, - width: 640, - height: 480, - } - } - }) + render! { + div { video { src: "test_video.mp4", autoplay: true, controls: true, width: 640, height: 480 } } + } } -fn get_stream_response( +async fn get_stream_response( + asset: &mut (impl tokio::io::AsyncSeek + tokio::io::AsyncRead + Unpin + Send + Sync), request: &AssetRequest, - boundary_id: &Arc>, ) -> Result>, Box> { - let video_file = PathBuf::from("./examples/assets/test_video.mp4"); - - let mut file = std::fs::File::open(&video_file)?; - - // get file length + // get stream length let len = { - let old_pos = file.stream_position()?; - let len = file.seek(SeekFrom::End(0))?; - file.seek(SeekFrom::Start(old_pos))?; + let old_pos = asset.stream_position().await?; + let len = asset.seek(SeekFrom::End(0)).await?; + asset.seek(SeekFrom::Start(old_pos)).await?; len }; @@ -103,9 +111,9 @@ fn get_stream_response( // allocate a buf with a suitable capacity let mut buf = Vec::with_capacity(bytes_to_read as usize); // seek the file to the starting byte - file.seek(SeekFrom::Start(start))?; + asset.seek(SeekFrom::Start(start)).await?; // read the needed bytes - file.take(bytes_to_read).read_to_end(&mut buf)?; + asset.take(bytes_to_read).read_to_end(&mut buf).await?; resp = resp.header(CONTENT_RANGE, format!("bytes {start}-{end}/{len}")); resp = resp.header(CONTENT_LENGTH, end + 1 - start); @@ -130,9 +138,7 @@ fn get_stream_response( }) .collect::>(); - let mut id = boundary_id.lock().unwrap(); - *id += 1; - let boundary = format!("sadasq2e{id}"); + let boundary = format!("{:x}", rand::random::()); let boundary_sep = format!("\r\n--{boundary}\r\n"); let boundary_closer = format!("\r\n--{boundary}\r\n"); @@ -143,34 +149,34 @@ fn get_stream_response( for (end, start) in ranges { // a new range is being written, write the range boundary - buf.write_all(boundary_sep.as_bytes())?; + buf.write_all(boundary_sep.as_bytes()).await?; // write the needed headers `Content-Type` and `Content-Range` - buf.write_all(format!("{CONTENT_TYPE}: video/mp4\r\n").as_bytes())?; - buf.write_all( - format!("{CONTENT_RANGE}: bytes {start}-{end}/{len}\r\n").as_bytes(), - )?; + buf.write_all(format!("{CONTENT_TYPE}: video/mp4\r\n").as_bytes()) + .await?; + buf.write_all(format!("{CONTENT_RANGE}: bytes {start}-{end}/{len}\r\n").as_bytes()) + .await?; // write the separator to indicate the start of the range body - buf.write_all("\r\n".as_bytes())?; + buf.write_all("\r\n".as_bytes()).await?; // calculate number of bytes needed to be read let bytes_to_read = end + 1 - start; let mut local_buf = vec![0_u8; bytes_to_read as usize]; - file.seek(SeekFrom::Start(start))?; - file.read_exact(&mut local_buf)?; + asset.seek(SeekFrom::Start(start)).await?; + asset.read_exact(&mut local_buf).await?; buf.extend_from_slice(&local_buf); } // all ranges have been written, write the closing boundary - buf.write_all(boundary_closer.as_bytes())?; + buf.write_all(boundary_closer.as_bytes()).await?; resp.body(buf) } } else { resp = resp.header(CONTENT_LENGTH, len); let mut buf = Vec::with_capacity(len as usize); - file.read_to_end(&mut buf)?; + asset.read_to_end(&mut buf).await?; resp.body(buf) }; diff --git a/packages/desktop/src/protocol.rs b/packages/desktop/src/protocol.rs index 31930da50..46be1b942 100644 --- a/packages/desktop/src/protocol.rs +++ b/packages/desktop/src/protocol.rs @@ -4,7 +4,7 @@ use slab::Slab; use std::{ borrow::Cow, future::Future, - ops::{Deref, DerefMut}, + ops::Deref, path::{Path, PathBuf}, pin::Pin, rc::Rc, @@ -72,10 +72,11 @@ pub type AssetResponse = Response>; pub trait AssetFuture: Future> + Send + Sync + 'static {} impl> + Send + Sync + 'static> AssetFuture for T {} +#[derive(Debug, Clone)] /// A request for an asset. This is a wrapper around [`Request>`] that provides methods specific to asset requests. pub struct AssetRequest { path: PathBuf, - request: Request>, + request: Arc>>, } impl AssetRequest { @@ -90,7 +91,10 @@ impl From>> for AssetRequest { let decoded = urlencoding::decode(request.uri().path().trim_start_matches('/')) .expect("expected URL to be UTF-8 encoded"); let path = PathBuf::from(&*decoded); - Self { request, path } + Self { + request: Arc::new(request), + path, + } } } @@ -102,12 +106,6 @@ impl Deref for AssetRequest { } } -impl DerefMut for AssetRequest { - fn deref_mut(&mut self) -> &mut Self::Target { - &mut self.request - } -} - /// A handler that takes an [`AssetRequest`] and returns a future that either loads the asset, or returns `None`. /// This handler is stashed indefinitely in a context object, so it must be `'static`. pub trait AssetHandler: Send + Sync + 'static {