Merge pull request #75 from FrancisRussell/develop

Release 0.6.0

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## 0.6.0
+
+* Add support for interrupting a stream rewrite.
+* Allow interrupts during stream rewrite in `opusgain`.
+* Add interrupt support to `opuscomment`.
+
 ## 0.5.1
 
 * Add Ctrl-C support for stopping `opusgain`.

Cargo.toml

@@ -1,6 +1,6 @@
 [package]
 name = "zoog"
-version = "0.5.1"
+version = "0.6.0"
 authors = ["Francis Russell <[email protected]>"]
 edition = "2018"
 homepage = "https://github.com/FrancisRussell/zoog"

src/bin/opuscomment.rs

@@ -1,5 +1,8 @@
 #![feature(let_chains)]
 
+#[path = "../ctrlc_handling.rs"]
+mod ctrlc_handling;
+
 #[path = "../output_file.rs"]
 mod output_file;
 
@@ -11,10 +14,11 @@ use std::ops::BitOrAssign;
 use std::path::{Path, PathBuf};
 
 use clap::Parser;
+use ctrlc_handling::CtrlCChecker;
 use output_file::OutputFile;
 use thiserror::Error;
 use zoog::comment_rewrite::{CommentHeaderRewrite, CommentRewriterAction, CommentRewriterConfig};
-use zoog::header_rewriter::{rewrite_stream, SubmitResult};
+use zoog::header_rewriter::{rewrite_stream_with_interrupt, SubmitResult};
 use zoog::opus::{parse_comment, validate_comment_field_name, CommentList, DiscreteCommentList};
 use zoog::{escaping, Error};
 
@@ -29,6 +33,9 @@ enum AppError {
     #[error("Silent exit because error was already printed")]
     SilentExit,
 
+    #[error("Unable to register Ctrl-C handler: `{0}`")]
+    CtrlCRegistration(#[from] ctrlc_handling::CtrlCRegistrationError),
+
     #[error("Failed to read from standard input: `{0}`")]
     StandardInputReadError(io::Error),
 }
@@ -245,6 +252,7 @@ fn read_comments_from_stdin(escaped: bool) -> Result<DiscreteCommentList, AppErr
 }
 
 fn main_impl() -> Result<(), AppError> {
+    let interrupt_checker = CtrlCChecker::new()?;
     let cli = Cli::parse_from(wild::args_os());
     let operation_mode = match (cli.list, cli.modify, cli.replace) {
         (_, false, false) => OperationMode::List,
@@ -302,7 +310,7 @@ fn main_impl() -> Result<(), AppError> {
         let mut output_file = BufWriter::new(output_file);
         let rewrite = CommentHeaderRewrite::new(rewriter_config);
         let abort_on_unchanged = true;
-        rewrite_stream(rewrite, &mut input_file, &mut output_file, abort_on_unchanged)
+        rewrite_stream_with_interrupt(rewrite, &mut input_file, &mut output_file, abort_on_unchanged, interrupt_checker)
     };
     drop(input_file); // Important for Windows
 

src/bin/opusgain.rs

@@ -3,8 +3,8 @@
 #[path = "../console_output.rs"]
 mod console_output;
 
-#[path = "../interrupt.rs"]
-mod interrupt;
+#[path = "../ctrlc_handling.rs"]
+mod ctrlc_handling;
 
 #[path = "../output_file.rs"]
 mod output_file;
@@ -17,14 +17,14 @@ use std::sync::atomic::{AtomicUsize, Ordering};
 
 use clap::{Parser, ValueEnum};
 use console_output::{ConsoleOutput, DelayedConsoleOutput, Standard};
-use interrupt::InterruptChecker;
+use ctrlc_handling::CtrlCChecker;
 use ogg::reading::PacketReader;
 use output_file::OutputFile;
 use parking_lot::Mutex;
 use rayon::iter::{IntoParallelIterator, ParallelIterator};
 use rayon::ThreadPoolBuilder;
 use thiserror::Error;
-use zoog::header_rewriter::{rewrite_stream, SubmitResult};
+use zoog::header_rewriter::{rewrite_stream_with_interrupt, SubmitResult};
 use zoog::opus::{TAG_ALBUM_GAIN, TAG_TRACK_GAIN};
 use zoog::volume_analyzer::VolumeAnalyzer;
 use zoog::volume_rewrite::{OpusGains, OutputGainMode, VolumeHeaderRewrite, VolumeRewriterConfig, VolumeTarget};
@@ -36,10 +36,7 @@ enum AppError {
     Library(#[from] Error),
 
     #[error("Unable to register Ctrl-C handler: `{0}`")]
-    InteruptRegistration(#[from] interrupt::InteruptRegistrationError),
-
-    #[error("Interrupted by user")]
-    UserInterrupted,
+    CtrlCRegistration(#[from] ctrlc_handling::CtrlCRegistrationError),
 }
 
 fn main() {
@@ -52,22 +49,22 @@ fn main() {
     }
 }
 
-fn check_running(checker: &InterruptChecker) -> Result<(), AppError> {
+fn check_running(checker: &CtrlCChecker) -> Result<(), Error> {
     if checker.is_running() {
         Ok(())
     } else {
-        Err(AppError::UserInterrupted)
+        Err(Error::Interrupted)
     }
 }
 
 fn apply_volume_analysis<P, C>(
-    analyzer: &mut VolumeAnalyzer, path: P, console_output: C, report_error: bool, interrupt_checker: InterruptChecker,
-) -> Result<(), AppError>
+    analyzer: &mut VolumeAnalyzer, path: P, console_output: C, report_error: bool, interrupt_checker: CtrlCChecker,
+) -> Result<(), Error>
 where
     P: AsRef<Path>,
     C: ConsoleOutput,
 {
-    let mut body = || -> Result<(), AppError> {
+    let mut body = || -> Result<(), Error> {
         let input_path = path.as_ref();
         let input_file = File::open(input_path).map_err(|e| Error::FileOpenError(input_path.to_path_buf(), e))?;
         let input_file = BufReader::new(input_file);
@@ -126,8 +123,8 @@ impl AlbumVolume {
 }
 
 fn compute_album_volume<I, P, C>(
-    paths: I, console_output: C, interrupt_checker: InterruptChecker,
-) -> Result<AlbumVolume, AppError>
+    paths: I, console_output: C, interrupt_checker: CtrlCChecker,
+) -> Result<AlbumVolume, Error>
 where
     I: IntoIterator<Item = P>,
     P: AsRef<Path>,
@@ -141,7 +138,7 @@ where
     // This is a BTreeMap so we process the analyzers in the supplied order
     let analyzers = Mutex::new(BTreeMap::new());
 
-    paths.into_par_iter().panic_fuse().try_for_each(|(idx, input_path)| -> Result<(), AppError> {
+    paths.into_par_iter().panic_fuse().try_for_each(|(idx, input_path)| -> Result<(), Error> {
         let mut analyzer = VolumeAnalyzer::default();
         apply_volume_analysis(
             &mut analyzer,
@@ -224,7 +221,7 @@ struct Cli {
 }
 
 fn main_impl() -> Result<(), AppError> {
-    let interrupt_checker = InterruptChecker::new()?;
+    let interrupt_checker = CtrlCChecker::new()?;
     let cli = Cli::parse_from(wild::args_os());
     let album_mode = cli.album;
     let num_threads = if cli.num_threads == 0 {
@@ -327,7 +324,13 @@ fn main_impl() -> Result<(), AppError> {
                     let mut output_file = BufWriter::new(output_file);
                     let rewrite = VolumeHeaderRewrite::new(rewriter_config);
                     let abort_on_unchanged = true;
-                    rewrite_stream(rewrite, &mut input_file, &mut output_file, abort_on_unchanged)
+                    rewrite_stream_with_interrupt(
+                        rewrite,
+                        &mut input_file,
+                        &mut output_file,
+                        abort_on_unchanged,
+                        interrupt_checker.clone(),
+                    )
                 };
                 drop(input_file); // Important for Windows
                 num_processed.fetch_add(1, Ordering::Relaxed);

src/ctrlc_handling.rs

@@ -0,0 +1,31 @@
+use std::sync::atomic::{AtomicBool, Ordering};
+use std::sync::Arc;
+
+use zoog::interrupt::Interrupt;
+
+pub type CtrlCRegistrationError = ctrlc::Error;
+
+#[derive(Clone, Debug)]
+pub struct CtrlCChecker {
+    running: Arc<AtomicBool>,
+}
+
+impl CtrlCChecker {
+    pub fn new() -> Result<CtrlCChecker, CtrlCRegistrationError> {
+        let running = Arc::new(AtomicBool::new(true));
+        {
+            let running = running.clone();
+            ctrlc::set_handler(move || {
+                running.store(false, Ordering::Relaxed);
+            })?;
+        }
+        let result = CtrlCChecker { running };
+        Ok(result)
+    }
+
+    pub fn is_running(&self) -> bool { self.running.load(Ordering::Relaxed) }
+}
+
+impl Interrupt for CtrlCChecker {
+    fn is_set(&self) -> bool { !self.is_running() }
+}

src/error.rs

@@ -100,4 +100,8 @@ pub enum Error {
     /// An escaped string was invalid
     #[error("{0}")]
     EscapeDecodeError(#[from] escaping::EscapeDecodeError),
+
+    /// An interrupt was detected
+    #[error("The operation was interrupted")]
+    Interrupted,
 }

src/header_rewriter.rs

@@ -5,6 +5,7 @@ use derivative::Derivative;
 use ogg::writing::{PacketWriteEndInfo, PacketWriter};
 use ogg::{Packet, PacketReader};
 
+use crate::interrupt::{Interrupt, Never};
 use crate::opus::{CommentHeader, OpusHeader};
 use crate::Error;
 
@@ -163,22 +164,31 @@ impl<HR: HeaderRewrite, W: Write> HeaderRewriter<'_, HR, W> {
     }
 }
 
-/// Convenience function for performing a rewrite. Rewrites the headers of an
-/// Ogg Opus stream using the supplied `HeaderRewrite`. If `abort_on_unchanged`
-/// is set, the function will terminate immediately if it is detected that no
-/// headers were modified, otherwise it will continue to rewrite the stream
-/// until the input stream is exhausted or an error occurs.
-pub fn rewrite_stream<HR: HeaderRewrite, R: Read + Seek, W: Write>(
-    rewrite: HR, input: R, mut output: W, abort_on_unchanged: bool,
+/// Convenience function for performing a rewrite.
+///
+/// Rewrites the headers of an Ogg Opus stream using the supplied
+/// `HeaderRewrite`. If `abort_on_unchanged` is set, the function will terminate
+/// immediately if it is detected that no headers were modified, otherwise it
+/// will continue to rewrite the stream until the input stream is exhausted, an
+/// error occurs or the interrupt condition is set.
+pub fn rewrite_stream_with_interrupt<HR, R, W, I>(
+    rewrite: HR, input: R, mut output: W, abort_on_unchanged: bool, interrupt: I,
 ) -> Result<SubmitResult<HR::Summary>, HR::Error>
 where
     HR::Error: From<Error>,
+    R: Read + Seek,
+    W: Write,
+    HR: HeaderRewrite,
+    I: Interrupt,
 {
     let mut ogg_reader = PacketReader::new(input);
     let ogg_writer = PacketWriter::new(&mut output);
     let mut rewriter = HeaderRewriter::new(rewrite, ogg_writer);
     let mut result = SubmitResult::Good;
     loop {
+        if interrupt.is_set() {
+            return Err(Error::Interrupted.into());
+        }
         match ogg_reader.read_packet() {
             Err(e) => break Err(Error::OggDecode(e).into()),
             Ok(None) => {
@@ -209,3 +219,17 @@ where
         }
     }
 }
+
+/// Identical to `rewrite_stream_with_interrupt` except the rewrite loop cannot
+/// be interrupted.
+pub fn rewrite_stream<HR, R, W>(
+    rewrite: HR, input: R, output: W, abort_on_unchanged: bool,
+) -> Result<SubmitResult<HR::Summary>, HR::Error>
+where
+    HR::Error: From<Error>,
+    R: Read + Seek,
+    W: Write,
+    HR: HeaderRewrite,
+{
+    rewrite_stream_with_interrupt(rewrite, input, output, abort_on_unchanged, Never::default())
+}

src/interrupt.rs

@@ -1,25 +1,13 @@
-use std::sync::atomic::{AtomicBool, Ordering};
-use std::sync::Arc;
-
-pub type InteruptRegistrationError = ctrlc::Error;
-
-#[derive(Clone, Debug)]
-pub struct InterruptChecker {
-    running: Arc<AtomicBool>,
+/// Allows reading the status of a potential interrupt
+pub trait Interrupt {
+    /// Has the interrupt been triggered?
+    fn is_set(&self) -> bool;
 }
 
-impl InterruptChecker {
-    pub fn new() -> Result<InterruptChecker, InteruptRegistrationError> {
-        let running = Arc::new(AtomicBool::new(true));
-        {
-            let running = running.clone();
-            ctrlc::set_handler(move || {
-                running.store(false, Ordering::Relaxed);
-            })?;
-        }
-        let result = InterruptChecker { running };
-        Ok(result)
-    }
+/// An interrupt that is never triggered
+#[derive(Debug, Default)]
+pub struct Never {}
 
-    pub fn is_running(&self) -> bool { self.running.load(Ordering::Relaxed) }
+impl Interrupt for Never {
+    fn is_set(&self) -> bool { false }
 }

src/lib.rs

@@ -18,6 +18,9 @@ pub mod header_rewriter;
 /// Functionality for rewriting Ogg Opus streams with new comments
 pub mod comment_rewrite;
 
+/// Support for detecting an operation should be interrupted
+pub mod interrupt;
+
 /// Functionality for rewriting Ogg Opus streams with altered output gain and
 /// volume tags
 pub mod volume_rewrite;