Clean dumper

CHANGELOG.md

@@ -20,6 +20,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   ([@mgrachev](https://github.com/mgrachev))
 
 ### ⚙️ Changed
+- Refactoring: generic indicator and writer [#132](https://github.com/datanymizer/datanymizer/pull/132)
+  ([@evgeniy-r](https://github.com/evgeniy-r))
 - Refactoring: remove unrelated code from Engine [#131](https://github.com/datanymizer/datanymizer/pull/131)
   ([@evgeniy-r](https://github.com/evgeniy-r))
 - Remove the PostgreSQL dependency from the CLI application [#123](https://github.com/datanymizer/datanymizer/pull/123)

cli/pg_datanymizer/src/app.rs

@@ -1,10 +1,12 @@
 use anyhow::Result;
+use std::{fs::File, io};
 use url::Url;
 
 use crate::options::{Options, TransactionConfig};
 
 use datanymizer_dumper::{
-    postgres::{connector::Connector, dumper::PgDumper, writer::DumpWriter, IsolationLevel},
+    indicator::{ConsoleIndicator, SilentIndicator},
+    postgres::{connector::Connector, dumper::PgDumper, IsolationLevel},
     Dumper,
 };
 use datanymizer_engine::{Engine, Settings};
@@ -25,10 +27,30 @@ impl App {
     }
 
     pub fn run(&self) -> Result<()> {
-        let mut dumper = self.dumper()?;
         let mut connection = self.connector().connect()?;
+        let engine = self.engine()?;
 
-        dumper.dump(&mut connection)
+        match &self.options.file {
+            Some(filename) => PgDumper::new(
+                engine,
+                self.dump_isolation_level(),
+                self.options.pg_dump_location.clone(),
+                File::create(filename)?,
+                ConsoleIndicator::new(),
+                self.options.pg_dump_args.clone(),
+            )?
+            .dump(&mut connection),
+
+            None => PgDumper::new(
+                engine,
+                self.dump_isolation_level(),
+                self.options.pg_dump_location.clone(),
+                io::stdout(),
+                SilentIndicator,
+                self.options.pg_dump_args.clone(),
+            )?
+            .dump(&mut connection),
+        }
     }
 
     fn connector(&self) -> Connector {
@@ -40,18 +62,6 @@ impl App {
         )
     }
 
-    fn dumper(&self) -> Result<PgDumper> {
-        let engine = self.engine()?;
-
-        PgDumper::new(
-            engine,
-            self.dump_isolation_level(),
-            self.options.pg_dump_location.clone(),
-            DumpWriter::new(self.options.file.clone())?,
-            self.options.pg_dump_args.clone(),
-        )
-    }
-
     fn engine(&self) -> Result<Engine> {
         let settings = Settings::new(self.options.config.clone())?;
         Ok(Engine::new(settings))

datanymizer_dumper/src/indicator.rs

@@ -0,0 +1,112 @@
+use indicatif::{HumanDuration, ProgressBar, ProgressStyle};
+use std::time::Duration;
+
+pub trait Indicator {
+    fn start_pb(&self, _size: u64, _prefix: &str) {}
+
+    fn inc_pb(&self, _i: u64) {}
+
+    fn finish_pb(&self, _name: &str, _duration: Duration) {}
+
+    fn debug_msg(&self, _msg: &str) {}
+}
+
+pub struct SilentIndicator;
+
+impl Indicator for SilentIndicator {}
+
+pub struct ConsoleIndicator {
+    pb: ProgressBar,
+}
+
+impl ConsoleIndicator {
+    pub fn new() -> Self {
+        Self::default()
+    }
+}
+
+impl Default for ConsoleIndicator {
+    fn default() -> Self {
+        let pb = ProgressBar::new(0);
+        Self { pb }
+    }
+}
+
+impl Indicator for ConsoleIndicator {
+    fn start_pb(&self, size: u64, name: &str) {
+        let delta = size / 100;
+        self.pb.set_length(size);
+        self.pb.set_draw_delta(delta);
+        self.pb.set_prefix(name);
+        self.pb.set_style(
+            ProgressStyle::default_bar()
+                .template(
+                    "[Dumping: {prefix}] [|{bar:50}|] {pos} of {len} rows [{percent}%] ({eta})",
+                )
+                .progress_chars("#>-"),
+        );
+    }
+
+    fn inc_pb(&self, i: u64) {
+        self.pb.inc(i);
+    }
+
+    fn finish_pb(&self, name: &str, duration: Duration) {
+        self.pb.finish();
+        self.pb.reset();
+
+        self.debug_msg(
+            format!(
+                "[Dumping: {}] Finished in {}",
+                name,
+                HumanDuration(duration)
+            )
+            .as_str(),
+        );
+    }
+
+    fn debug_msg(&self, msg: &str) {
+        println!("{}", msg);
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    // just test that there is no panic
+    mod console_indicator {
+        use super::*;
+
+        #[test]
+        fn debug_msg() {
+            ConsoleIndicator::new().debug_msg("some message");
+        }
+
+        #[test]
+        fn pb_start_finish() {
+            let ci = ConsoleIndicator::new();
+            ci.start_pb(100, "name");
+            ci.finish_pb("name", Duration::new(1, 0));
+        }
+
+        #[test]
+        fn pb_some_progress() {
+            let ci = ConsoleIndicator::new();
+            ci.start_pb(100, "name");
+            ci.inc_pb(1);
+            ci.inc_pb(10);
+            ci.finish_pb("name", Duration::new(1, 0));
+        }
+
+        #[test]
+        fn pb_overflow_progress() {
+            let ci = ConsoleIndicator::new();
+            ci.start_pb(100, "name");
+            ci.inc_pb(1);
+            ci.inc_pb(10);
+            ci.inc_pb(100);
+            ci.finish_pb("name", Duration::new(1, 0));
+        }
+    }
+}

datanymizer_dumper/src/lib.rs

@@ -5,13 +5,14 @@ use indicatif::HumanDuration;
 use solvent::DepGraph;
 use std::{collections::HashMap, hash::Hash, time::Instant};
 
+pub mod indicator;
 pub mod postgres;
 
 // Dumper makes dump with same stages
 pub trait Dumper: 'static + Sized + Send {
     type Table;
     type Connection;
-    type SchemaInspector: SchemaInspector<Dumper = Self>;
+    type SchemaInspector: SchemaInspector<Connection = Self::Connection>;
 
     /// Process steps
     fn dump(&mut self, connection: &mut Self::Connection) -> Result<()> {
@@ -52,34 +53,25 @@ pub trait Dumper: 'static + Sized + Send {
 
 pub trait SchemaInspector: 'static + Sized + Send + Clone {
     type Type;
-    type Dumper: Dumper;
+    type Connection;
     type Table: Table<Self::Type>;
     type Column: ColumnData<Self::Type>;
 
     /// Get all tables in the database
-    fn get_tables(
-        &self,
-        connection: &mut <Self::Dumper as Dumper>::Connection,
-    ) -> Result<Vec<Self::Table>>;
+    fn get_tables(&self, connection: &mut Self::Connection) -> Result<Vec<Self::Table>>;
 
     /// Get table size
-    fn get_table_size(
-        &self,
-        connection: &mut <Self::Dumper as Dumper>::Connection,
-        table: &Self::Table,
-    ) -> Result<i64>;
+    fn get_table_size(&self, connection: &mut Self::Connection, table: &Self::Table)
+        -> Result<i64>;
 
     /// Get all dependencies (by FK) for `table` in database
     fn get_dependencies(
         &self,
-        connection: &mut <Self::Dumper as Dumper>::Connection,
+        connection: &mut Self::Connection,
         table: &Self::Table,
     ) -> Result<Vec<Self::Table>>;
 
-    fn ordered_tables(
-        &self,
-        connection: &mut <Self::Dumper as Dumper>::Connection,
-    ) -> Vec<(Self::Table, i32)> {
+    fn ordered_tables(&self, connection: &mut Self::Connection) -> Vec<(Self::Table, i32)> {
         let mut res: HashMap<Self::Table, i32> = HashMap::new();
         let mut depgraph: DepGraph<Self::Table> = DepGraph::new();
         if let Ok(tables) = self.get_tables(connection) {
@@ -108,14 +100,13 @@ pub trait SchemaInspector: 'static + Sized + Send + Clone {
     /// Get columns for table
     fn get_columns(
         &self,
-        connection: &mut <Self::Dumper as Dumper>::Connection,
+        connection: &mut Self::Connection,
         table: &Self::Table,
     ) -> Result<Vec<Self::Column>>;
 }
 
 /// Table trait for all databases
 pub trait Table<T>: Sized + Send + Clone + Eq + Hash {
-    type Dumper: Dumper;
     type Column: ColumnData<T>;
     type Row;