Fileio Support [CPP-183]

console_backend/Cargo.toml

@@ -16,8 +16,8 @@ capnp = "0.14"
 chrono = { version = "0.4", features = ["serde"] }
 csv = "1"
 paste = "1"
-pyo3 = { version = "0.13", features = ["extension-module"] }
-sbp = { git = "https://github.com/swift-nav/libsbp.git", rev = "a5c221684439f2124306a2daa44126aac5457c80", features = ["json", "swiftnav-rs"] }
+pyo3 = { version = "0.13", features = ["extension-module"], optional = true }
+sbp = { git = "https://github.com/swift-nav/libsbp.git", rev = "ce5e3d14ae4284c53982bc3fd79a2fa15976ff9f", features = ["json", "swiftnav-rs"] }
 serde = { version = "1.0.123", features = ["derive"] }
 tempfile = "3.2.0"
 ordered-float = "2.0"
@@ -33,6 +33,8 @@ clap = "3.0.0-beta.2"
 indexmap = { version = "1.6.2", features = ["serde"] }
 serde_json = { version = "1" }
 crossbeam = "0.8.1"
+rand = "0.8.3"
+slotmap = "1.0.3"
 
 [target.'cfg(any(target_os = "macos", target_os = "windows"))'.dependencies]
 serialport = "4.0.1"
@@ -53,7 +55,12 @@ name = "console_backend"
 crate-type = ["cdylib", "lib"]
 bench = false
 
+[[bin]]
+name = "fileio"
+path = "src/bin/fileio.rs"
+bench = false
+
 [features]
-default = []
+default = ["pyo3"]
 benches = []
 tests = []

console_backend/benches/cpu_benches.rs

@@ -3,7 +3,7 @@ use criterion::{criterion_group, criterion_main, Criterion};
 use glob::glob;
 use sbp::sbp_tools::SBPTools;
 use std::{
-    fs,
+    fs, io,
     path::Path,
     sync::{mpsc, Arc, Mutex},
     thread, time,
@@ -12,7 +12,7 @@ use std::{
 extern crate console_backend;
 use console_backend::{
     process_messages,
-    types::{ClientSender, RealtimeDelay, SharedState},
+    types::{ClientSender, Connection, RealtimeDelay, SharedState},
 };
 
 const BENCH_FILEPATH: &str = "./tests/data/piksi-relay.sbp";
@@ -61,15 +61,8 @@ fn run_process_messages(file_in_name: &str, failure: bool) {
             inner: client_send_,
         };
         shared_state.set_running(true, client_send.clone());
-        match fs::File::open(file_in_name) {
-            Ok(fileopen) => process_messages::process_messages(
-                fileopen,
-                shared_state,
-                client_send,
-                RealtimeDelay::Off,
-            ),
-            Err(e) => panic!("unable to read file, {}.", e),
-        }
+        let conn = Connection::file(file_in_name.into()).unwrap();
+        process_messages::process_messages(conn, shared_state, client_send, RealtimeDelay::Off);
     }
     recv_thread.join().expect("join should succeed");
 }

console_backend/src/bin/fileio.rs

@@ -0,0 +1,133 @@
+use std::{
+    fs,
+    io::{self, Write},
+};
+
+use clap::Clap;
+use crossbeam::{channel, scope};
+use sbp::sbp_tools::SBPTools;
+
+use console_backend::{
+    broadcaster::Broadcaster,
+    cli_options::Input,
+    fileio::Fileio,
+    types::{MsgSender, Result},
+};
+
+#[derive(Clap, Debug)]
+#[clap(about = "Fileio operations.")]
+pub enum Opts {
+    /// Write a file from local source to remote destination dest.
+    Write {
+        source: String,
+        dest: String,
+        #[clap(subcommand)]
+        input: Input,
+    },
+
+    /// Read a file from remote source to local dest. If no dest is provided, file is read to stdout.
+    Read {
+        source: String,
+        dest: Option<String>,
+        #[clap(subcommand)]
+        input: Input,
+    },
+
+    /// List a directory.
+    List {
+        path: String,
+        #[clap(subcommand)]
+        input: Input,
+    },
+
+    /// Delete a file.
+    Delete {
+        path: String,
+        #[clap(subcommand)]
+        input: Input,
+    },
+}
+
+fn main() -> Result<()> {
+    let bc = Broadcaster::new();
+
+    let (done_tx, done_rx) = channel::bounded(0);
+
+    let bc_source = bc.clone();
+    let run = move |rdr| {
+        let messages = sbp::iter_messages(rdr).log_errors(log::Level::Debug);
+        for msg in messages {
+            bc_source.send(&msg);
+            if done_rx.try_recv().is_ok() {
+                break;
+            }
+        }
+    };
+
+    match Opts::parse() {
+        Opts::Write {
+            source,
+            dest,
+            input,
+        } => {
+            let (rdr, wtr) = input.into_conn()?.into_io();
+            let sender = MsgSender::new(wtr);
+            scope(|s| {
+                s.spawn(|_| run(rdr));
+                let mut fileio = Fileio::new(bc, sender);
+                let data = fs::File::open(source)?;
+                fileio.overwrite(dest, data)?;
+                eprintln!("file written successfully.");
+                done_tx.send(true).unwrap();
+                Result::Ok(())
+            })
+            .unwrap()
+        }
+        Opts::Read {
+            source,
+            dest,
+            input,
+        } => {
+            let (rdr, wtr) = input.into_conn()?.into_io();
+            let sender = MsgSender::new(wtr);
+            scope(|s| {
+                s.spawn(|_| run(rdr));
+                let mut fileio = Fileio::new(bc, sender);
+                let dest: Box<dyn Write> = match dest {
+                    Some(path) => Box::new(fs::File::create(path)?),
+                    None => Box::new(io::stdout()),
+                };
+                fileio.read(source, dest)?;
+                done_tx.send(true).unwrap();
+                Result::Ok(())
+            })
+            .unwrap()
+        }
+        Opts::List { path, input } => {
+            let (rdr, wtr) = input.into_conn()?.into_io();
+            let sender = MsgSender::new(wtr);
+            scope(|s| {
+                s.spawn(|_| run(rdr));
+                let mut fileio = Fileio::new(bc, sender);
+                let files = fileio.readdir(path)?;
+                eprintln!("{:#?}", files);
+                done_tx.send(true).unwrap();
+                Result::Ok(())
+            })
+            .unwrap()
+        }
+        Opts::Delete { path, input } => {
+            let (rdr, wtr) = input.into_conn()?.into_io();
+            let sender = MsgSender::new(wtr);
+            scope(|s| {
+                s.spawn(|_| run(rdr));
+                let fileio = Fileio::new(bc, sender);
+                fileio.remove(path)?;
+                eprintln!("file deleted.");
+                done_tx.send(true).unwrap();
+                Result::Ok(())
+            })
+            .unwrap()
+        }
+    }
+}