chore: Rename compression directory to compression_engine

src/compression_engine/gzip.rs

@@ -0,0 +1,74 @@
+use anyhow::Result;
+use log::*;
+use std::fs::File;
+use std::io;
+use std::io::{Read, Write};
+use std::path::PathBuf;
+
+use flate2::read::GzDecoder;
+use flate2::write::GzEncoder;
+use flate2::Compression;
+
+use crate::compression::CompressionEngine;
+
+#[derive(Debug, Eq, PartialEq, Clone, Default)]
+pub struct CompressionEngineGZip {}
+
+impl CompressionEngineGZip {
+    pub fn new() -> CompressionEngineGZip {
+        CompressionEngineGZip {}
+    }
+}
+
+impl CompressionEngine for CompressionEngineGZip {
+    fn is_supported(&self) -> bool {
+        true
+    }
+
+    fn open(&self, file_path: PathBuf) -> Result<Box<dyn Read>> {
+        debug!("COMPRESSION: Opening {:?} using {:?}", file_path, *self);
+
+        let file = File::open(file_path)?;
+        Ok(Box::new(GzDecoder::new(file)))
+    }
+
+    fn create(&self, file_path: PathBuf) -> Result<Box<dyn Write>> {
+        debug!("COMPRESSION: Writting to {:?} using {:?}", file_path, *self);
+
+        let file = File::create(file_path)?;
+        let gzip_write = GzEncoder::new(file, Compression::default());
+
+        Ok(Box::new(AutoFinishGzEncoder::new(gzip_write)))
+    }
+}
+
+pub struct AutoFinishGzEncoder<W: Write> {
+    encoder: Option<GzEncoder<W>>,
+}
+
+impl<W: Write> AutoFinishGzEncoder<W> {
+    fn new(gz_encoder: GzEncoder<W>) -> AutoFinishGzEncoder<W> {
+        AutoFinishGzEncoder {
+            encoder: Some(gz_encoder),
+        }
+    }
+}
+
+impl<W: Write> Drop for AutoFinishGzEncoder<W> {
+    fn drop(&mut self) {
+        if let Some(encoder) = self.encoder.take() {
+            debug!("COMPRESSION: Finishing");
+            let _ = encoder.finish();
+        }
+    }
+}
+
+impl<W: Write> Write for AutoFinishGzEncoder<W> {
+    fn write(&mut self, buf: &[u8]) -> io::Result<usize> {
+        self.encoder.as_mut().unwrap().write(buf)
+    }
+
+    fn flush(&mut self) -> io::Result<()> {
+        self.encoder.as_mut().unwrap().flush()
+    }
+}

src/compression_engine/lz4.rs

@@ -0,0 +1,40 @@
+use anyhow::Result;
+use log::*;
+use std::fs::File;
+use std::io::{Read, Write};
+use std::path::PathBuf;
+
+use lz4_flex::frame::{FrameDecoder, FrameEncoder};
+
+use crate::compression::CompressionEngine;
+
+#[derive(Debug, Eq, PartialEq, Clone, Default)]
+pub struct CompressionEngineLZ4 {}
+
+impl CompressionEngineLZ4 {
+    pub fn new() -> CompressionEngineLZ4 {
+        CompressionEngineLZ4 {}
+    }
+}
+
+impl CompressionEngine for CompressionEngineLZ4 {
+    fn is_supported(&self) -> bool {
+        true
+    }
+
+    fn open(&self, file_path: PathBuf) -> Result<Box<dyn Read>> {
+        debug!("COMPRESSION: Opening {:?} using {:?}", file_path, *self);
+
+        let file = File::open(file_path)?;
+        Ok(Box::new(FrameDecoder::new(file)))
+    }
+
+    fn create(&self, file_path: PathBuf) -> Result<Box<dyn Write>> {
+        debug!("COMPRESSION: Writting to {:?} using {:?}", file_path, *self);
+
+        let file = File::create(file_path)?;
+        let lz4_write = FrameEncoder::new(file).auto_finish();
+
+        Ok(Box::new(lz4_write))
+    }
+}

src/compression_engine/none.rs

@@ -0,0 +1,33 @@
+use anyhow::Result;
+use log::*;
+use std::fs::File;
+use std::io::{Read, Write};
+use std::path::PathBuf;
+
+use crate::compression::CompressionEngine;
+
+#[derive(Debug, Eq, PartialEq, Clone, Default)]
+pub struct CompressionEngineNone {}
+
+impl CompressionEngineNone {
+    pub fn new() -> CompressionEngineNone {
+        CompressionEngineNone {}
+    }
+}
+
+impl CompressionEngine for CompressionEngineNone {
+    fn size(&self, file_path: PathBuf) -> Result<usize> {
+        let item_file_metadata = file_path.metadata()?;
+        Ok(item_file_metadata.len() as usize)
+    }
+
+    fn open(&self, file_path: PathBuf) -> Result<Box<dyn Read>> {
+        debug!("COMPRESSION: Opening {:?} using {:?}", file_path, *self);
+        Ok(Box::new(File::open(file_path)?))
+    }
+
+    fn create(&self, file_path: PathBuf) -> Result<Box<dyn Write>> {
+        debug!("COMPRESSION: Writting to {:?} using {:?}", file_path, *self);
+        Ok(Box::new(File::create(file_path)?))
+    }
+}

src/compression_engine/program.rs

@@ -0,0 +1,114 @@
+use anyhow::{anyhow, Context, Result};
+use log::*;
+use std::env;
+use std::fs;
+use std::fs::File;
+use std::io::{Read, Write};
+use std::os::unix::fs::PermissionsExt;
+use std::path::PathBuf;
+use std::process::{Command, Stdio};
+
+use crate::compression::CompressionEngine;
+
+#[derive(Debug, Eq, PartialEq, Clone)]
+pub struct CompressionEngineProgram {
+    pub program: String,
+    pub compress: Vec<String>,
+    pub decompress: Vec<String>,
+    pub supported: bool,
+}
+
+impl CompressionEngineProgram {
+    pub fn new(
+        program: &str,
+        compress: Vec<&str>,
+        decompress: Vec<&str>,
+    ) -> CompressionEngineProgram {
+        let program_path = get_program_path(program);
+        let supported = program_path.is_ok();
+
+        CompressionEngineProgram {
+            program: program_path.unwrap_or(program.to_string()),
+            compress: compress.iter().map(|s| s.to_string()).collect(),
+            decompress: decompress.iter().map(|s| s.to_string()).collect(),
+            supported,
+        }
+    }
+}
+
+impl CompressionEngine for CompressionEngineProgram {
+    fn is_supported(&self) -> bool {
+        self.supported
+    }
+
+    fn open(&self, file_path: PathBuf) -> Result<Box<dyn Read>> {
+        debug!("COMPRESSION: Opening {:?} using {:?}", file_path, *self);
+
+        let program = self.program.clone();
+        let args = self.decompress.clone();
+
+        debug!(
+            "COMPRESSION: Executing command: {:?} {:?} reading from {:?}",
+            program, args, file_path
+        );
+
+        let file = File::open(file_path).context("Unable to open file for reading")?;
+
+        let process = Command::new(program.clone())
+            .args(args.clone())
+            .stdin(file)
+            .stdout(Stdio::piped())
+            .spawn()
+            .context(anyhow!(
+                "Unable to spawn child process: {:?} {:?}",
+                program,
+                args
+            ))?;
+        Ok(Box::new(process.stdout.unwrap()))
+    }
+
+    fn create(&self, file_path: PathBuf) -> Result<Box<dyn Write>> {
+        debug!("COMPRESSION: Writting to {:?} using {:?}", file_path, *self);
+
+        let program = self.program.clone();
+        let args = self.compress.clone();
+
+        debug!(
+            "COMPRESSION: Executing command: {:?} {:?} writing to {:?}",
+            program, args, file_path
+        );
+
+        let file = File::create(file_path).context("Unable to open file for writing")?;
+
+        let process = Command::new(program.clone())
+            .args(args.clone())
+            .stdin(Stdio::piped())
+            .stdout(file)
+            .spawn()
+            .context(anyhow!(
+                "Problem spawning child process: {:?} {:?}",
+                program,
+                args
+            ))?;
+
+        Ok(Box::new(process.stdin.unwrap()))
+    }
+}
+
+fn get_program_path(program: &str) -> Result<String> {
+    debug!("COMPRESSION: Looking for executable: {}", program);
+    if let Ok(path) = env::var("PATH") {
+        for p in path.split(':') {
+            let p_str = format!("{}/{}", p, program);
+            let stat = fs::metadata(p_str.clone());
+            if let Ok(stat) = stat {
+                let md = stat;
+                let permissions = md.permissions();
+                if md.is_file() && permissions.mode() & 0o111 != 0 {
+                    return Ok(p_str);
+                }
+            }
+        }
+    }
+    Err(anyhow!("Unable to find binary {} in PATH", program))
+}