code cleanups

Author: mangelajo

Cargo.lock

@@ -78,7 +78,7 @@ pub(crate) struct BlockWriter {
     buffer_pos: usize, // Current position in buffer
     bytes_written: u64,
     bytes_since_sync: u64, // Track bytes written since last sync
-    written_tx: mpsc::UnboundedSender<u64>,
+    written_progress_tx: mpsc::UnboundedSender<u64>,
     use_direct_io: bool, // Track if O_DIRECT is active
     #[allow(dead_code)]
     debug: bool, // Debug mode flag
@@ -91,7 +91,7 @@ impl BlockWriter {
     /// Open a block device for writing with direct I/O
     pub(crate) fn new(
         device: &str,
-        written_tx: mpsc::UnboundedSender<u64>,
+        written_progress_tx: mpsc::UnboundedSender<u64>,
         debug: bool,
         o_direct: bool,
     ) -> io::Result<Self> {
@@ -222,7 +222,7 @@ impl BlockWriter {
             buffer_pos: 0,
             bytes_written: 0,
             bytes_since_sync: 0,
-            written_tx,
+            written_progress_tx,
             use_direct_io,
             debug,
         })
@@ -254,7 +254,7 @@ impl BlockWriter {
 
             // Send progress update periodically (every 256KB to reduce overhead)
             if self.bytes_written.is_multiple_of(256 * 1024) {
-                let _ = self.written_tx.send(self.bytes_written);
+                let _ = self.written_progress_tx.send(self.bytes_written);
             }
         }
 
@@ -304,7 +304,7 @@ impl BlockWriter {
         self.file.sync_all()?;
         self.bytes_since_sync = 0;
         // Send final progress update
-        let _ = self.written_tx.send(self.bytes_written);
+        let _ = self.written_progress_tx.send(self.bytes_written);
         Ok(())
     }
 
@@ -324,7 +324,7 @@ impl AsyncBlockWriter {
     /// Create a new async block writer
     pub(crate) fn new(
         device: String,
-        written_tx: mpsc::UnboundedSender<u64>,
+        written_progress_tx: mpsc::UnboundedSender<u64>,
         debug: bool,
         o_direct: bool,
     ) -> io::Result<Self> {
@@ -333,7 +333,7 @@ impl AsyncBlockWriter {
         // Spawn blocking task for I/O operations
         let writer_handle = tokio::task::spawn_blocking(move || {
             let mut writer =
-                BlockWriter::new(&device, written_tx, debug, o_direct).map_err(|e| {
+                BlockWriter::new(&device, written_progress_tx, debug, o_direct).map_err(|e| {
                     eprintln!("Failed to open device '{}': {}", device, e);
                     e
                 })?;

src/fls/block_writer.rs

@@ -1,4 +1,4 @@
-use tokio::io::{AsyncReadExt, AsyncWriteExt};
+use tokio::io::AsyncReadExt;
 use tokio::process::{Child, Command};
 use tokio::sync::mpsc;
 
@@ -49,38 +49,6 @@ pub(crate) async fn start_decompressor_process(
     Ok((process, cmd))
 }
 
-#[allow(dead_code)]
-pub(crate) async fn spawn_decompressor_stdout_reader(
-    mut stdout: tokio::process::ChildStdout,
-    decompressed_tx: mpsc::UnboundedSender<Vec<u8>>,
-    mut dd_stdin: tokio::process::ChildStdin,
-    error_tx: mpsc::UnboundedSender<String>,
-) {
-    let mut buffer = [0u8; 8 * 1024 * 1024]; // 8MB buffer for better performance
-    loop {
-        match stdout.read(&mut buffer).await {
-            Ok(0) => break, // EOF
-            Ok(n) => {
-                // Send to progress tracking
-                if decompressed_tx.send(buffer[..n].to_vec()).is_err() {
-                    break;
-                }
-                // Write to dd stdin
-                if let Err(e) = dd_stdin.write_all(&buffer[..n]).await {
-                    let _ = error_tx.send(format!("Error writing to dd stdin: {}", e));
-                    break;
-                }
-            }
-            Err(e) => {
-                let _ = error_tx.send(format!("Error reading from decompressor stdout: {}", e));
-                break;
-            }
-        }
-    }
-    // Close dd stdin when decompressor is done
-    let _ = dd_stdin.shutdown().await;
-}
-
 pub(crate) async fn spawn_stderr_reader(
     mut stderr: tokio::process::ChildStderr,
     error_tx: mpsc::UnboundedSender<String>,

src/fls/decompress.rs

@@ -24,16 +24,16 @@ pub async fn flash_from_url(
     let decompressor_stderr = decompressor.stderr.take().unwrap();
 
     // Create channels
-    let (decompressed_tx, mut decompressed_rx) = mpsc::unbounded_channel::<u64>();
+    let (decompressed_progress_tx, mut decompressed_progress_rx) = mpsc::unbounded_channel::<u64>();
     let (error_tx, error_rx) = mpsc::unbounded_channel::<String>();
-    let (written_tx, mut written_rx) = mpsc::unbounded_channel::<u64>();
+    let (written_progress_tx, mut written_progress_rx) = mpsc::unbounded_channel::<u64>();
 
     println!("Opening block device for writing: {}", options.device);
 
     // Create block writer
     let block_writer = AsyncBlockWriter::new(
         options.device.clone(),
-        written_tx,
+        written_progress_tx,
         options.debug,
         options.o_direct,
     )?;
@@ -52,7 +52,7 @@ pub async fn flash_from_url(
                     Ok(n) => {
                         let data = buffer[..n].to_vec();
                         // Send byte count to progress tracking
-                        if decompressed_tx.send(n as u64).is_err() {
+                        if decompressed_progress_tx.send(n as u64).is_err() {
                             break;
                         }
                         // Write to block device
@@ -111,7 +111,7 @@ pub async fn flash_from_url(
     let (buffer_tx, mut buffer_rx) = mpsc::channel::<bytes::Bytes>(buffer_capacity);
 
     // Channels for tracking bytes actually written to decompressor
-    let (decompressor_written_tx, mut decompressor_written_rx) = mpsc::unbounded_channel::<u64>();
+    let (decompressor_written_progress_tx, mut decompressor_written_progress_rx) = mpsc::unbounded_channel::<u64>();
 
     // Spawn persistent task to write buffered chunks to decompressor
     let decompressor_writer_handle = tokio::spawn(async move {
@@ -121,7 +121,7 @@ pub async fn flash_from_url(
                 return Err(format!("Error writing to decompressor stdin: {}", e));
             }
             // Notify that bytes were written to decompressor
-            let _ = decompressor_written_tx.send(chunk_len);
+            let _ = decompressor_written_progress_tx.send(chunk_len);
         }
         // Close decompressor stdin when channel is closed
         Ok::<(), String>(())
@@ -220,7 +220,7 @@ pub async fn flash_from_url(
                             retry_count = 0; // Reset retry count on successful download
 
                             // Track bytes actually written to decompressor
-                            while let Ok(written_len) = decompressor_written_rx.try_recv() {
+                            while let Ok(written_len) = decompressor_written_progress_rx.try_recv() {
                                 bytes_sent_to_decompressor += written_len;
                                 progress.bytes_sent_to_decompressor += written_len;
                             }
@@ -234,11 +234,11 @@ pub async fn flash_from_url(
                             }
 
                             // Update progress from other channels
-                            while let Ok(byte_count) = decompressed_rx.try_recv() {
+                            while let Ok(byte_count) = decompressed_progress_rx.try_recv() {
                                 progress.bytes_decompressed += byte_count;
                             }
 
-                            while let Ok(written_bytes) = written_rx.try_recv() {
+                            while let Ok(written_bytes) = written_progress_rx.try_recv() {
                                 progress.bytes_written = written_bytes;
                             }
 
@@ -338,18 +338,18 @@ pub async fn flash_from_url(
         // Update progress from all channels
         let mut updated = false;
 
-        while let Ok(written_len) = decompressor_written_rx.try_recv() {
+        while let Ok(written_len) = decompressor_written_progress_rx.try_recv() {
             bytes_sent_to_decompressor += written_len;
             progress.bytes_sent_to_decompressor += written_len;
             updated = true;
         }
 
-        while let Ok(byte_count) = decompressed_rx.try_recv() {
+        while let Ok(byte_count) = decompressed_progress_rx.try_recv() {
             progress.bytes_decompressed += byte_count;
             updated = true;
         }
 
-        while let Ok(written_bytes) = written_rx.try_recv() {
+        while let Ok(written_bytes) = written_progress_rx.try_recv() {
             progress.bytes_written = written_bytes;
             updated = true;
         }
@@ -381,7 +381,7 @@ pub async fn flash_from_url(
     }
 
     // Update any remaining progress
-    while let Ok(byte_count) = decompressed_rx.try_recv() {
+    while let Ok(byte_count) = decompressed_progress_rx.try_recv() {
         progress.bytes_decompressed += byte_count;
     }
 
@@ -413,12 +413,12 @@ pub async fn flash_from_url(
             // Update progress from channels
             let mut updated = false;
 
-            while let Ok(byte_count) = decompressed_rx.try_recv() {
+            while let Ok(byte_count) = decompressed_progress_rx.try_recv() {
                 progress.bytes_decompressed += byte_count;
                 updated = true;
             }
 
-            while let Ok(written_bytes) = written_rx.try_recv() {
+            while let Ok(written_bytes) = written_progress_rx.try_recv() {
                 progress.bytes_written = written_bytes;
                 updated = true;
             }
@@ -466,7 +466,7 @@ pub async fn flash_from_url(
         // Update progress from channels
         let mut updated = false;
 
-        while let Ok(written_bytes) = written_rx.try_recv() {
+        while let Ok(written_bytes) = written_progress_rx.try_recv() {
             progress.bytes_written = written_bytes;
             updated = true;
         }
@@ -508,11 +508,11 @@ pub async fn flash_from_url(
     }
 
     // Read any remaining progress updates
-    while let Ok(byte_count) = decompressed_rx.try_recv() {
+    while let Ok(byte_count) = decompressed_progress_rx.try_recv() {
         progress.bytes_decompressed += byte_count;
     }
 
-    while let Ok(written_bytes) = written_rx.try_recv() {
+    while let Ok(written_bytes) = written_progress_rx.try_recv() {
         progress.bytes_written = written_bytes;
     }