navactor 0.1.3

A cli tool for creating and updating actors from piped input
Documentation 
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122

use clap::{Args, Parser, Subcommand};
use navactor::director;
use navactor::json_decoder;
use navactor::message::Message;
use navactor::message::Message::EndOfStream;
use navactor::stdin_actor;
use navactor::stdout_actor;
use navactor::store_actor_sqlite;
use tokio::runtime::Runtime;

#[derive(Parser)]
#[command(
    author,
    version,
    about,
    long_about = "nv is the CLI for the DtLaboratory project"
)] // Read from `Cargo.toml`
#[command(propagate_version = true)]
struct Cli {
    #[arg(short, long)]
    dbfile: Option<String>,
    #[arg(short, long)]
    buffer: Option<usize>,
    #[arg(short, long, action = clap::ArgAction::Count)]
    verbose: u8,
    #[command(subcommand)]
    command: Commands,
}

#[derive(Subcommand)]
enum Commands {
    /// Process incoming stream of internal formats into a namespace.  A
    /// namespace is an instance of GraphDirector.
    Update(Namespace),
    /// Get actor state for all actors in path
    Inspect(NvPath),
}

#[derive(Args)]
struct Namespace {
    namespace: String,
}

#[derive(Args)]
struct NoArg {}

#[derive(Args)]
struct NvPath {
    /// actor path
    path: String,
}

#[derive(Args)]
struct Extractor {
    /// name of extractor
    extractor: String,
}

fn update(namespace: Namespace, bufsz: usize, runtime: Runtime) {
    let result = run_async_update(namespace, bufsz);
    runtime.block_on(result).expect("An error occurred")
}

async fn run_async_update(namespace: Namespace, bufsz: usize) -> Result<(), String> {
    let output = stdout_actor::new(bufsz); // print state changes
    let store_actor = store_actor_sqlite::new(bufsz, namespace.namespace.clone()); // print state changes
    let director_w_persist =
        director::new(namespace.namespace, bufsz, Some(output), Some(store_actor));
    let json_decoder_actor = json_decoder::new(bufsz, director_w_persist); // parse input
    let input = stdin_actor::new(bufsz, json_decoder_actor); // read from stdin
    match input.ask(Message::ReadAllCmd {}).await {
        EndOfStream {} => Ok(()),
        _ => Err("END and response: sucks.".to_string()),
    }
}

fn inspect(path: NvPath, bufsz: usize, runtime: Runtime) {
    let result = run_async_inspect(path, bufsz);
    runtime.block_on(result).expect("An error occurred");
}

async fn run_async_inspect(path: NvPath, bufsz: usize) -> Result<(), String> {
    let p = std::path::Path::new(&path.path);
    let ns = p
        .components()
        .find(|c| c == &std::path::Component::RootDir)
        .unwrap()
        .as_os_str()
        .to_str()
        .unwrap();
    //let ns = p.iter().next().unwrap().to_str().unwrap();
    log::trace!("inspect of ns {}", ns);
    let output = stdout_actor::new(bufsz); // print state
    let store_actor = store_actor_sqlite::new(bufsz, String::from(ns)); // print state
    let director = director::new(path.path.clone(), bufsz, None, Some(store_actor));
    let m = director.ask(Message::Query { path: path.path }).await;
    output.tell(m).await;
    // send complete to keep the job running long enough to print the above
    match output.ask(Message::EndOfStream {}).await {
        EndOfStream {} => Ok(()),
        _ => Err("END and response: sucks.".to_string()),
    }
}

/// control logging of nv and various libs via RUST_LOG env var like so:
///std::env::set_var("RUST_LOG", "debug,sqlx=warn");
fn main() {
    env_logger::init();
    log::info!("nv started");

    let cli = Cli::parse();
    let bufsz: usize = cli.buffer.unwrap_or(8);

    let runtime = Runtime::new().unwrap_or_else(|e| panic!("Error creating runtime: {}", e));

    match cli.command {
        Commands::Update(namespace) => update(namespace, bufsz, runtime),
        Commands::Inspect(path) => inspect(path, bufsz, runtime),
    }

    log::info!("nv stopped");
}