fixup! Refactor JPEG handling in agent

agent/src/ffmpeg/streamer.rs

@@ -19,15 +19,12 @@ use tracing::{debug, error, info, warn};
 
 const HLS_PLAYLIST_FILENAME: &str = "stream.m3u8";
 
-type JpegSender = broadcast::Sender<bytes::Bytes>;
-type JpegReceiver = broadcast::Receiver<bytes::Bytes>;
-
 pub(crate) struct Streamer {
     config: Config,
     terminate: Arc<Mutex<bool>>,
     ffmpeg_pid: Arc<Mutex<Option<Pid>>>,
     handle: Option<JoinHandle<()>>,
-    jpeg_tx: JpegSender,
+    jpeg_tx: broadcast::Sender<bytes::Bytes>,
 }
 
 impl Streamer {
@@ -43,7 +40,7 @@ impl Streamer {
         }
     }
 
-    pub(crate) fn jpeg_subscribe(&self) -> JpegReceiver {
+    pub(crate) fn jpeg_subscribe(&self) -> broadcast::Receiver<bytes::Bytes> {
         self.jpeg_tx.subscribe()
     }
 

agent/src/jpeg_frame_decoder.rs

@@ -1,21 +1,18 @@
 use bytes::{Buf, Bytes, BytesMut};
 use tokio_util::codec::Decoder;
 
-pub(crate) struct JpegFrameDecoder;
+const JPEG_EOI_LENGTH: usize = 2;
 
-impl JpegFrameDecoder {
-    pub(crate) fn default() -> Self {
-        Self {}
-    }
-}
+#[derive(Default)]
+pub(crate) struct JpegFrameDecoder;
 
 impl Decoder for JpegFrameDecoder {
     type Item = Bytes;
     type Error = std::io::Error;
 
     fn decode(&mut self, buf: &mut BytesMut) -> Result<Option<Self::Item>, Self::Error> {
         if let Some(idx) = find_first_jpeg_eoi(buf) {
-            let image_buf = buf.copy_to_bytes(idx + 2);
+            let image_buf = buf.copy_to_bytes(idx + JPEG_EOI_LENGTH);
             Ok(Some(image_buf))
         } else {
             Ok(None)
@@ -24,11 +21,11 @@ impl Decoder for JpegFrameDecoder {
 }
 
 fn find_first_jpeg_eoi(bytes: &BytesMut) -> Option<usize> {
-    if bytes.len() < 2 {
-        return None;
+    if bytes.len() < JPEG_EOI_LENGTH {
+        None
+    } else {
+        (0..bytes.len() - 1).find(|&i| bytes[i] == 0xFF && bytes[i + 1] == 0xD9)
     }
-
-    (0..bytes.len() - 1).find(|&i| bytes[i] == 0xFF && bytes[i + 1] == 0xD9)
 }
 
 #[cfg(test)]