perf(video): 优化 MJPEG 传输和 JPEG 编码

按 HTTP 消费进度读取最新帧并合并积压通知，复用帧缓冲区所有权，减少 MJPEG 传输和截图的数据复制。

在 JPEG 数据后立即发送下一帧的完整 multipart 头，减少静态画面的浏览器显示等待，并保留保活和客户端释放逻辑。

缓存相同原始帧的 JPEG 编码结果，在输入或画质变化时重新编码；按像素格式分配转换缓冲区，减少闲置内存。

补充缓冲区寿命、慢客户端、序列号重置、编码缓存和保活回归测试。

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<RawFrameFingerprint>,
+    last_frame: Option<VideoFrame>,
+}
+
 #[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<Option<JpegEncoder>>,
+    jpeg_encoder: ParkingMutex<Option<JpegEncoderState>>,
     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) => {
+            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
             );
-                    enc
-                }
-                Err(e) => {
-                    warn!("Failed to create JPEG encoder: {}", e);
-                    panic!("Failed to create JPEG encoder");
-                }
-            }
+            *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();

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<MjpegStreamHandler>,
+    guard: ClientGuard,
+) -> impl Stream<Item = Result<Bytes, io::Error>> + Send {
+    stream_with_timeout(handler, guard, Duration::from_secs(5))
+}
+
+fn stream_with_timeout(
+    handler: Arc<MjpegStreamHandler>,
+    guard: ClientGuard,
+    timeout: Duration,
+) -> impl Stream<Item = Result<Bytes, io::Error>> + 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);
+    }
+}

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;
 

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<Self> {
-        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<Self> {
         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());
     }
 }

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::<bytes::Bytes>(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::<bytes::Bytes, std::io::Error>(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<Arc<AppState>>) -> 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<Arc<AppState>>) -> 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()
-}