From 92a08fb2bc5122cca0db22a247a9676999e1c7ac Mon Sep 17 00:00:00 2001 From: mofeng-git Date: Sun, 18 Oct 2026 17:42:22 +0800 Subject: [PATCH] =?UTF-8?q?perf(video):=20=E4=BC=98=E5=8C=96=20MJPEG=20?= =?UTF-8?q?=E4=BC=A0=E8=BE=93=E5=92=8C=20JPEG=20=E7=BC=96=E7=A0=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 HTTP 消费进度读取最新帧并合并积压通知，复用帧缓冲区所有权，减少 MJPEG 传输和截图的数据复制。 在 JPEG 数据后立即发送下一帧的完整 multipart 头，减少静态画面的浏览器显示等待，并保留保活和客户端释放逻辑。 缓存相同原始帧的 JPEG 编码结果，在输入或画质变化时重新编码；按像素格式分配转换缓冲区，减少闲置内存。 补充缓冲区寿命、慢客户端、序列号重置、编码缓存和保活回归测试。 --- src/stream/mjpeg.rs | 166 ++++++++++++++++++++++++++++----- src/stream/mjpeg_http.rs | 184 +++++++++++++++++++++++++++++++++++++ src/stream/mod.rs | 1 + src/video/codec/jpeg.rs | 60 ++++++++---- src/web/handlers/stream.rs | 139 +--------------------------- 5 files changed, 376 insertions(+), 174 deletions(-) create mode 100644 src/stream/mjpeg_http.rs diff --git a/src/stream/mjpeg.rs b/src/stream/mjpeg.rs index e6713d28..a016f752 100644 --- a/src/stream/mjpeg.rs +++ b/src/stream/mjpeg.rs @@ -24,6 +24,25 @@ use crate::video::VideoFrame; pub type ClientId = String; +#[cfg(feature = "desktop")] +#[derive(Clone, Copy, PartialEq, Eq)] +struct RawFrameFingerprint { + resolution: crate::video::format::Resolution, + format: PixelFormat, + stride: u32, + online: bool, + len: usize, + hash: u64, + quality: u32, +} + +#[cfg(feature = "desktop")] +struct JpegEncoderState { + encoder: JpegEncoder, + last_input: Option, + last_frame: Option, +} + #[derive(Debug, Clone)] pub struct ClientSession { pub id: ClientId, @@ -115,7 +134,7 @@ pub struct MjpegStreamHandler { dropped_same_frames: AtomicU64, max_drop_same_frames: AtomicU64, #[cfg(feature = "desktop")] - jpeg_encoder: ParkingMutex>, + jpeg_encoder: ParkingMutex>, jpeg_quality: AtomicU64, } @@ -219,24 +238,43 @@ impl MjpegStreamHandler { let resolution = frame.resolution; let sequence = self.sequence.load(Ordering::Relaxed); let desired_quality = self.jpeg_quality.load(Ordering::Relaxed) as u32; + // Keep only a fingerprint and the encoded output, not a second raw + // frame. Unchanged input can reuse JPEG bytes even for forced refreshes. + let fingerprint = RawFrameFingerprint { + resolution, + format: frame.format, + stride: frame.stride, + online: frame.online, + len: frame.len(), + hash: frame.get_hash(), + quality: desired_quality, + }; let mut encoder_guard = self.jpeg_encoder.lock(); - let encoder = encoder_guard.get_or_insert_with(|| { + if encoder_guard.is_none() { let config = EncoderConfig::jpeg(resolution, desired_quality); - match JpegEncoder::new(config) { - Ok(enc) => { - debug!( - "Created JPEG encoder for MJPEG stream: {}x{} (q={})", - resolution.width, resolution.height, desired_quality - ); - enc - } - Err(e) => { - warn!("Failed to create JPEG encoder: {}", e); - panic!("Failed to create JPEG encoder"); - } + let encoder = JpegEncoder::new(config) + .map_err(|e| format!("Failed to create JPEG encoder: {}", e))?; + debug!( + "Created JPEG encoder for MJPEG stream: {}x{} (q={})", + resolution.width, resolution.height, desired_quality + ); + *encoder_guard = Some(JpegEncoderState { + encoder, + last_input: None, + last_frame: None, + }); + } + let state = encoder_guard.as_mut().expect("JPEG encoder initialized"); + if state.last_input == Some(fingerprint) { + if let Some(mut cached) = state.last_frame.clone() { + cached.sequence = sequence; + cached.capture_ts = frame.capture_ts; + cached.online = frame.online; + return Ok(cached); } - }); + } + let encoder = &mut state.encoder; if encoder.config().resolution != resolution { debug!( @@ -285,13 +323,12 @@ impl MjpegStreamHandler { } }; - Ok(VideoFrame::new( - encoded.data, - resolution, - PixelFormat::Mjpeg, - 0, - sequence, - )) + let mut jpeg = VideoFrame::new(encoded.data, resolution, PixelFormat::Mjpeg, 0, sequence); + jpeg.capture_ts = frame.capture_ts; + jpeg.online = frame.online; + state.last_input = Some(fingerprint); + state.last_frame = Some(jpeg.clone()); + Ok(jpeg) } pub fn set_offline(&self) { @@ -550,6 +587,91 @@ mod tests { use crate::video::{format::Resolution, PixelFormat}; use bytes::Bytes; + #[cfg(feature = "desktop")] + fn raw_frame(resolution: Resolution, sequence: u64) -> VideoFrame { + VideoFrame::new( + Bytes::from(vec![ + 128; + resolution.width as usize + * resolution.height as usize + * 3 + / 2 + ]), + resolution, + PixelFormat::Nv12, + resolution.width, + sequence, + ) + } + + #[cfg(feature = "desktop")] + #[test] + fn raw_cache_reuses_bytes_but_invalidates_on_input_and_quality_changes() { + let handler = Arc::new(MjpegStreamHandler::with_drop_limit(0)); + let _guard = ClientGuard::new("raw-test".into(), handler.clone()); + let resolution = Resolution::new(64, 32); + handler.update_frame(raw_frame(resolution, 1)); + let first = handler.current_frame().unwrap(); + let next_input = raw_frame(resolution, 2); + let capture_ts = next_input.capture_ts; + handler.update_frame(next_input); + let repeated = handler.current_frame().unwrap(); + assert_eq!(first.data().as_ptr(), repeated.data().as_ptr()); + assert_ne!(first.sequence, repeated.sequence); + assert_eq!(repeated.capture_ts, capture_ts); + + handler.set_jpeg_quality(40); + handler.update_frame(raw_frame(resolution, 3)); + let changed_quality = handler.current_frame().unwrap(); + assert_ne!(changed_quality.data().as_ptr(), repeated.data().as_ptr()); + let mut data = raw_frame(resolution, 4).data().to_vec(); + data[321] = 255; // A small change away from the fast sample positions. + handler.update_frame(VideoFrame::new( + Bytes::from(data), + resolution, + PixelFormat::Nv12, + resolution.width, + 4, + )); + let changed_pixels = handler.current_frame().unwrap(); + assert_ne!( + changed_pixels.data().as_ptr(), + changed_quality.data().as_ptr() + ); + + let resized = Resolution::new(32, 64); + handler.update_frame(raw_frame(resized, 5)); + let changed_resolution = handler.current_frame().unwrap(); + assert_eq!(changed_resolution.resolution, resized); + assert_ne!( + changed_resolution.data().as_ptr(), + changed_pixels.data().as_ptr() + ); + } + + #[cfg(feature = "desktop")] + #[test] + fn raw_duplicates_skip_publication_but_forced_refresh_reuses_jpeg() { + let handler = Arc::new(MjpegStreamHandler::new()); + let _guard = ClientGuard::new("raw-refresh-test".into(), handler.clone()); + let mut notifications = handler.subscribe(); + let resolution = Resolution::new(64, 32); + handler.update_frame(raw_frame(resolution, 1)); + let first = handler.current_frame().unwrap(); + notifications.try_recv().unwrap(); + handler.update_frame(raw_frame(resolution, 2)); + assert!(matches!( + notifications.try_recv(), + Err(broadcast::error::TryRecvError::Empty) + )); + *handler.last_frame_ts.write() = Some(Instant::now() - Duration::from_secs(2)); + handler.update_frame(raw_frame(resolution, 3)); + notifications.try_recv().unwrap(); + let refresh = handler.current_frame().unwrap(); + assert_ne!(refresh.sequence, first.sequence); + assert_eq!(refresh.data().as_ptr(), first.data().as_ptr()); + } + #[tokio::test] async fn test_stream_handler() { let handler = MjpegStreamHandler::new(); diff --git a/src/stream/mjpeg_http.rs b/src/stream/mjpeg_http.rs new file mode 100644 index 00000000..7070d887 --- /dev/null +++ b/src/stream/mjpeg_http.rs @@ -0,0 +1,184 @@ +//! Pull the latest JPEG directly when the HTTP consumer is ready. + +use std::io; +use std::sync::Arc; +use std::time::Duration; + +use bytes::Bytes; +use futures::Stream; +use tokio::sync::broadcast::error::{RecvError, TryRecvError}; + +use super::{ClientGuard, MjpegStreamHandler}; + +const PART_HEADER: &[u8] = b"--frame\r\nContent-Type: image/jpeg\r\n\r\n"; +// Chromium commits the previous image after parsing the next part's complete +// headers. Send these immediately after the JPEG, even when the scene stops. +const PART_TRAILER: &[u8] = b"\r\n--frame\r\nContent-Type: image/jpeg\r\n\r\n"; + +pub(crate) fn body_stream( + handler: Arc, + guard: ClientGuard, +) -> impl Stream> + Send { + stream_with_timeout(handler, guard, Duration::from_secs(5)) +} + +fn stream_with_timeout( + handler: Arc, + guard: ClientGuard, + timeout: Duration, +) -> impl Stream> + Send { + // Subscribe before reading the initial frame so an update cannot be missed. + let mut notifications = handler.subscribe(); + async_stream::stream! { + let guard = guard; + let mut last_sequence = None; + let mut timeout_count = 0; + let mut keepalive = false; + + yield Ok(Bytes::from_static(PART_HEADER)); + loop { + // Notifications contain no frame data. Coalesce the backlog, then + // fetch the latest frame rather than queueing JPEGs for slow peers. + loop { + match notifications.try_recv() { + Ok(()) | Err(TryRecvError::Lagged(_)) => timeout_count = 0, + Err(TryRecvError::Empty) => break, + Err(TryRecvError::Closed) => return, + } + } + if !handler.is_online() { + break; + } + + if let Some(frame) = handler.current_frame() { + if (keepalive || last_sequence != Some(frame.sequence)) && frame.is_valid_jpeg() { + last_sequence = Some(frame.sequence); + // This retains the pooled owner; it does not copy the JPEG. + let data = frame.owned_data_bytes(); + drop(frame); + handler.record_frame_sent(guard.id()); + yield Ok(data); + yield Ok(Bytes::from_static(PART_TRAILER)); + } + } + + keepalive = false; + match tokio::time::timeout(timeout, notifications.recv()).await { + Ok(Ok(())) | Ok(Err(RecvError::Lagged(_))) => timeout_count = 0, + Ok(Err(RecvError::Closed)) => break, + Err(_) => { + timeout_count += 1; + if timeout_count > 6 { + break; + } + keepalive = true; + } + } + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::video::format::{PixelFormat, Resolution}; + use crate::video::frame::{FrameBuffer, FrameBufferPool, VideoFrame}; + use futures::StreamExt; + + fn jpeg(sequence: u64) -> VideoFrame { + let mut data = vec![sequence as u8; 128]; + data[..2].copy_from_slice(&[0xff, 0xd8]); + data[126..].copy_from_slice(&[0xff, 0xd9]); + VideoFrame::new( + Bytes::from(data), + Resolution::VGA, + PixelFormat::Mjpeg, + 0, + sequence, + ) + } + + #[tokio::test] + async fn sends_complete_pooled_frame_without_copying_or_waiting_for_next_frame() { + let handler = Arc::new(MjpegStreamHandler::with_drop_limit(0)); + let pool = Arc::new(FrameBufferPool::new(1)); + let data = jpeg(0).data().to_vec(); + let original = data.as_ptr(); + handler.update_frame(VideoFrame::from_pooled( + Arc::new(FrameBuffer::new(data, Some(pool.clone()))), + Resolution::VGA, + PixelFormat::Mjpeg, + 0, + 0, + )); + let guard = ClientGuard::new("http-test".into(), handler.clone()); + let mut stream = Box::pin(body_stream(handler.clone(), guard)); + assert_eq!(stream.next().await.unwrap().unwrap().as_ref(), PART_HEADER); + let payload = stream.next().await.unwrap().unwrap(); + assert_eq!(payload.as_ptr(), original); + assert_eq!(stream.next().await.unwrap().unwrap().as_ref(), PART_TRAILER); + assert!( + tokio::time::timeout(Duration::from_millis(20), stream.next()) + .await + .is_err() + ); + + // Exceed the broadcast capacity while the HTTP consumer is stalled. + for sequence in 1..=40 { + handler.update_frame(jpeg(sequence)); + } + let latest = stream.next().await.unwrap().unwrap(); + assert_eq!(latest[2], 40); + assert_eq!(stream.next().await.unwrap().unwrap().as_ref(), PART_TRAILER); + // The previous pooled frame remains valid until its HTTP owner drops. + assert_eq!(payload.as_ptr(), original); + assert_eq!(payload[2], 0); + drop(payload); + let recycled = pool.take(128); + assert_eq!(recycled.as_ptr(), original); + + // Capture sequence numbers may reset after reopening the source. + handler.update_frame(jpeg(0)); + assert_eq!(stream.next().await.unwrap().unwrap()[2], 0); + assert_eq!(stream.next().await.unwrap().unwrap().as_ref(), PART_TRAILER); + assert_eq!(handler.get_clients_stat()["http-test"].fps, 3); + drop(stream); + assert_eq!(handler.client_count(), 0); + } + + #[tokio::test] + async fn keepalive_and_offline_close_release_the_client() { + let handler = Arc::new(MjpegStreamHandler::new()); + handler.update_frame(jpeg(1)); + let guard = ClientGuard::new("keepalive-test".into(), handler.clone()); + let mut stream = Box::pin(stream_with_timeout( + handler.clone(), + guard, + Duration::from_millis(10), + )); + for _ in 0..3 { + stream.next().await.unwrap().unwrap(); + } + let repeated = tokio::time::timeout(Duration::from_secs(1), stream.next()) + .await + .unwrap() + .unwrap() + .unwrap(); + assert_eq!(repeated[2], 1); + assert_eq!(stream.next().await.unwrap().unwrap().as_ref(), PART_TRAILER); + handler.set_offline(); + assert!(stream.next().await.is_none()); + drop(stream); + assert_eq!(handler.client_count(), 0); + } + + #[test] + fn dropping_an_unpolled_response_releases_the_client() { + let handler = Arc::new(MjpegStreamHandler::new()); + let guard = ClientGuard::new("unpolled-test".into(), handler.clone()); + let stream = body_stream(handler.clone(), guard); + assert_eq!(handler.client_count(), 1); + drop(stream); + assert_eq!(handler.client_count(), 0); + } +} diff --git a/src/stream/mod.rs b/src/stream/mod.rs index b1413490..495c49f6 100644 --- a/src/stream/mod.rs +++ b/src/stream/mod.rs @@ -1,6 +1,7 @@ //! MJPEG multipart streaming and WebSocket HID (for MJPEG mode). pub mod mjpeg; +pub(crate) mod mjpeg_http; #[cfg(feature = "desktop")] pub mod ws_hid; diff --git a/src/video/codec/jpeg.rs b/src/video/codec/jpeg.rs index 3c2fe79a..03f3182f 100644 --- a/src/video/codec/jpeg.rs +++ b/src/video/codec/jpeg.rs @@ -33,14 +33,6 @@ pub struct JpegEncoder { impl JpegEncoder { /// Create a new JPEG encoder pub fn new(config: EncoderConfig) -> Result { - let resolution = config.resolution; - let width = resolution.width as usize; - let height = resolution.height as usize; - // I420: Y = width*height, U = width*height/4, V = width*height/4 - let i420_size = width * height * 3 / 2; - let max_uv_plane_size = width * height; - let bgra_size = width * height * 4; - let mut compressor = turbojpeg::Compressor::new().map_err(|e| { AppError::VideoError(format!("Failed to create turbojpeg compressor: {}", e)) })?; @@ -52,12 +44,18 @@ impl JpegEncoder { Ok(Self { config, compressor, - i420_buffer: vec![0u8; i420_size], - uv_split_buffer: vec![0u8; max_uv_plane_size * 2], - bgra_buffer: vec![0u8; bgra_size], + i420_buffer: Vec::new(), + uv_split_buffer: Vec::new(), + bgra_buffer: Vec::new(), }) } + fn ensure_i420_buffer(&mut self) { + let size = + self.config.resolution.width as usize * self.config.resolution.height as usize * 3 / 2; + self.i420_buffer.resize(size, 0); + } + /// Create with specific quality pub fn with_quality(resolution: Resolution, quality: u32) -> Result { let config = EncoderConfig::jpeg(resolution, quality); @@ -146,6 +144,7 @@ impl JpegEncoder { } // Convert YUYV to I420 using libyuv (SIMD accelerated) + self.ensure_i420_buffer(); libyuv::yuy2_to_i420(data, &mut self.i420_buffer, width as i32, height as i32) .map_err(|e| AppError::VideoError(format!("libyuv YUYV→I420 failed: {}", e)))?; @@ -166,6 +165,7 @@ impl JpegEncoder { ))); } + self.ensure_i420_buffer(); // Reuse bgra_buffer as scratch for the swapped YUYV data. if self.bgra_buffer.len() < expected_size { self.bgra_buffer.resize(expected_size, 0); @@ -202,6 +202,7 @@ impl JpegEncoder { } // Convert NV12 to I420 using libyuv (SIMD accelerated) + self.ensure_i420_buffer(); libyuv::nv12_to_i420(data, &mut self.i420_buffer, width as i32, height as i32) .map_err(|e| AppError::VideoError(format!("libyuv NV12→I420 failed: {}", e)))?; @@ -224,6 +225,8 @@ impl JpegEncoder { ))); } + self.ensure_i420_buffer(); + self.uv_split_buffer.resize(y_size, 0); let src_uv = &data[y_size..expected_size]; let chroma_plane_size = y_size / 2; let (u_plane_422, rest) = self.uv_split_buffer.split_at_mut(chroma_plane_size); @@ -273,6 +276,8 @@ impl JpegEncoder { ))); } + self.uv_split_buffer.resize(y_size * 2, 0); + self.bgra_buffer.resize(y_size * 4, 0); let src_uv = &data[y_size..expected_size]; let chroma_plane_size = y_size; let (u_plane_444, rest) = self.uv_split_buffer.split_at_mut(chroma_plane_size); @@ -318,6 +323,7 @@ impl JpegEncoder { } // Convert RGB24 to I420 using libyuv (SIMD accelerated) + self.ensure_i420_buffer(); libyuv::rgb24_to_i420(data, &mut self.i420_buffer, width as i32, height as i32) .map_err(|e| AppError::VideoError(format!("libyuv RGB24→I420 failed: {}", e)))?; @@ -340,6 +346,7 @@ impl JpegEncoder { // Convert BGR24 to I420 using libyuv (SIMD accelerated) // Note: libyuv's RAWToI420 is BGR24 → I420 + self.ensure_i420_buffer(); libyuv::bgr24_to_i420(data, &mut self.i420_buffer, width as i32, height as i32) .map_err(|e| AppError::VideoError(format!("libyuv BGR24→I420 failed: {}", e)))?; @@ -395,10 +402,31 @@ mod tests { use super::*; #[test] - fn test_i420_buffer_size() { - // 1920x1080 I420 = 1920*1080 + 960*540 + 960*540 = 3110400 bytes - let config = EncoderConfig::jpeg(Resolution::HD1080, 80); - let encoder = JpegEncoder::new(config).unwrap(); - assert_eq!(encoder.i420_buffer.len(), 1920 * 1080 * 3 / 2); + fn scratch_buffers_are_allocated_only_for_the_formats_that_use_them() { + let resolution = Resolution::new(64, 32); + let pixels = 64 * 32; + let mut encoder = JpegEncoder::new(EncoderConfig::jpeg(resolution, 80)).unwrap(); + assert!(encoder.i420_buffer.is_empty()); + assert!(encoder.uv_split_buffer.is_empty()); + assert!(encoder.bgra_buffer.is_empty()); + encoder.encode_yuyv(&vec![128; pixels * 2], 0).unwrap(); + encoder.encode_nv12(&vec![128; pixels * 3 / 2], 1).unwrap(); + encoder.encode_rgb(&vec![128; pixels * 3], 2).unwrap(); + encoder.encode_bgr(&vec![128; pixels * 3], 3).unwrap(); + assert_eq!(encoder.i420_buffer.len(), pixels * 3 / 2); + assert!(encoder.uv_split_buffer.is_empty()); + assert!(encoder.bgra_buffer.is_empty()); + + encoder.encode_yvyu(&vec![128; pixels * 2], 4).unwrap(); + assert_eq!(encoder.bgra_buffer.len(), pixels * 2); + encoder.encode_nv16(&vec![128; pixels * 2], 5).unwrap(); + assert_eq!(encoder.uv_split_buffer.len(), pixels); + encoder.encode_nv24(&vec![128; pixels * 3], 6).unwrap(); + assert_eq!(encoder.uv_split_buffer.len(), pixels * 2); + assert_eq!(encoder.bgra_buffer.len(), pixels * 4); + + let mut nv24 = JpegEncoder::new(EncoderConfig::jpeg(resolution, 80)).unwrap(); + nv24.encode_nv24(&vec![128; pixels * 3], 0).unwrap(); + assert!(nv24.i420_buffer.is_empty()); } } diff --git a/src/web/handlers/stream.rs b/src/web/handlers/stream.rs index 6f1743de..d70b818b 100644 --- a/src/web/handlers/stream.rs +++ b/src/web/handlers/stream.rs @@ -470,122 +470,8 @@ pub async fn mjpeg_stream( .filter(|id| !id.is_empty() && id.len() <= 64) // Validate: non-empty, max 64 chars .unwrap_or_else(|| uuid::Uuid::new_v4().to_string()); - // Create RAII guard - this will automatically register and unregister the client - let guard = Arc::new(crate::stream::mjpeg::ClientGuard::new( - client_id.clone(), - handler.clone(), - )); - - let (tx, mut rx) = tokio::sync::mpsc::channel::(1); - - let guard_clone = guard.clone(); - let handler_clone = handler.clone(); - tokio::spawn(async move { - let _guard = guard_clone; // Keep guard alive - let mut notify_rx = handler_clone.subscribe(); - let mut last_seq = 0u64; - let mut timeout_count = 0; - - // Send initial frame if available - if let Some(frame) = handler_clone.current_frame() { - if frame.is_valid_jpeg() { - let data = create_mjpeg_part(frame.data()); - // send() blocks until receiver is ready (backpressure) - if tx.send(data).await.is_ok() { - // FPS recording moved to async_stream after yield - last_seq = frame.sequence; - } else { - return; // Receiver closed - } - } - } - - loop { - // Check if stream went offline (e.g., during config change) - if !handler_clone.is_online() { - break; - } - - // Wait for new frame notification with timeout - let result = - tokio::time::timeout(std::time::Duration::from_secs(5), notify_rx.recv()).await; - - match result { - Ok(Ok(())) => { - // Check online status after receiving notification - // set_offline() sends a notification, so we need to check here - if !handler_clone.is_online() { - break; - } - timeout_count = 0; - if let Some(frame) = handler_clone.current_frame() { - // Use != instead of > to handle sequence reset when capturer restarts - // (e.g., after video config change, new capturer starts from seq=0) - if frame.sequence != last_seq && frame.is_valid_jpeg() { - let data = create_mjpeg_part(frame.data()); - if tx.send(data).await.is_ok() { - last_seq = frame.sequence; - } else { - break; - } - } - } - } - Ok(Err(tokio::sync::broadcast::error::RecvError::Closed)) => { - break; - } - Ok(Err(tokio::sync::broadcast::error::RecvError::Lagged(_))) => { - // Receiver was too slow - skip missed frames and jump to latest - if !handler_clone.is_online() { - break; - } - timeout_count = 0; - - if let Some(frame) = handler_clone.current_frame() { - if frame.is_valid_jpeg() { - // Send current frame immediately and reset sequence tracking - if tx.send(create_mjpeg_part(frame.data())).await.is_ok() { - last_seq = frame.sequence; - } else { - break; - } - } - } - } - Err(_) => { - // Timeout - check if still online - timeout_count += 1; - if timeout_count > 6 || !handler_clone.is_online() { - break; - } - // Send last frame again to keep connection alive - let Some(frame) = handler_clone.current_frame() else { - continue; - }; - - if frame.is_valid_jpeg() - && tx.send(create_mjpeg_part(frame.data())).await.is_err() - { - break; - } - } - } - } - }); - - // Create stream that receives from channel and forwards to the HTTP - // body. Record FPS *before* yield so the final frame of a session - // still gets counted (after-yield code in async_stream! only runs - // when the consumer polls again, which never happens for the last - // frame of a closing connection). - let handler_for_stream = handler.clone(); - let guard_for_stream = guard.clone(); - let body_stream = async_stream::stream! { - while let Some(data) = rx.recv().await { - handler_for_stream.record_frame_sent(guard_for_stream.id()); - yield Ok::(data); - } - }; + let guard = crate::stream::mjpeg::ClientGuard::new(client_id, handler.clone()); + let body_stream = crate::stream::mjpeg_http::body_stream(handler, guard); Response::builder() .status(StatusCode::OK) @@ -610,7 +496,7 @@ pub async fn snapshot(State(state): State>) -> impl IntoResponse { .status(StatusCode::OK) .header(header::CONTENT_TYPE, "image/jpeg") .header(header::CACHE_CONTROL, "no-cache") - .body(Body::from(frame.data_bytes())) + .body(Body::from(frame.owned_data_bytes())) .unwrap(), _ => Response::builder() .status(StatusCode::SERVICE_UNAVAILABLE) @@ -618,22 +504,3 @@ pub async fn snapshot(State(state): State>) -> impl IntoResponse { .unwrap(), } } - -/// Create MJPEG multipart frame bytes -fn create_mjpeg_part(jpeg_data: &[u8]) -> bytes::Bytes { - use bytes::{BufMut, BytesMut}; - - let mut buf = BytesMut::with_capacity(128 + jpeg_data.len()); - - // Write boundary and headers - buf.put_slice(b"--frame\r\n"); - buf.put_slice(b"Content-Type: image/jpeg\r\n"); - buf.put_slice(format!("Content-Length: {}\r\n", jpeg_data.len()).as_bytes()); - buf.put_slice(b"\r\n"); - - // Write JPEG data - buf.put_slice(jpeg_data); - buf.put_slice(b"\r\n"); - - buf.freeze() -}