valve-rs 0.1.2

Auto-scale R {plumber} APIs concurrently using Tokio, Axum, and Deadpool
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

use std::process::Stdio;
use valve::start;

use {argh::FromArgs, std::fmt::Debug};
#[derive(FromArgs, Debug)]
/// Distribute your plumber API in parallel.
struct Cli {
    /// host to serve APIs on
    #[argh(option, short = 'h', default = r#"String::from("127.0.0.1")"#)]
    host: String,

    /// the port to serve the main application on
    #[argh(option, short = 'p', default = "3000")]
    port: u16,

    /// the maximum number of plumber APIs to spawn
    #[argh(option, short = 'n', default = "3")]
    n_max: u16,

    /// number of Tokio workers to spawn to handle requests
    #[argh(option, short = 'w', default = "3")]
    workers: u16,

    /// path to the plumber API (default `plumber.R`)
    #[argh(option, short = 'f', default = r#"String::from("plumber.R")"#)]
    file: String,

    /// default 10. Interval in seconds when to check for unused connections
    #[argh(option, default = "10")]
    check_unused: u32,

    /// default 5 mins. How long an API can go unused before being killed in seconds.
    #[argh(option, default = "300")]
    max_age: u32,

    /// the maximum number of plumber APIs to spawn
    #[argh(option, default = "1")]
    n_min: u16,
}

//#[tokio::main(worker_threads = 5)]
fn main() {
    let cli_args: Cli = argh::from_env();

    // validate that the file exists
    let p = std::path::Path::new(&cli_args.file).try_exists().unwrap();
    if !p {
        panic!("plumber file does not exist.")
    }

    if cli_args.n_min < 1 {
        panic!("Cannot have fewer than 1 plumber API")
    }

    if cli_args.n_max < 1 {
        panic!("Cannot have fewer than 1 plumber API")
    }

    if cli_args.n_min > cli_args.n_max {
        panic!("`n_min` cannot be greater than `n_max`")
    }

    if cli_args.workers < 1 {
        panic!("Cannot have fewer than 1 worker thread")
    }

    // verify that R is installed will panic if R isn't found (i hope)
    std::process::Command::new("R")
        .arg("--version")
        .stdout(Stdio::null())
        .spawn()
        .unwrap();

    // verify that the plumber pcackage can be found
    let plumber_exists = std::process::Command::new("Rscript")
        .arg("-e")
        .arg("library(plumber)")
        .stderr(Stdio::piped())
        .output()
        .unwrap()
        .status
        .success();

    // panic if plumber isn't found
    if !plumber_exists {
        panic!("plumber package cannot be found")
    }

    // build the tokio runtime
    tokio::runtime::Builder::new_multi_thread()
        .worker_threads(cli_args.workers as usize)
        .enable_all()
        .build()
        .unwrap()
        .block_on(async {
            println!("Valve starting at: {}:{}", cli_args.host, cli_args.port);
            start::valve_start(
                cli_args.file,
                cli_args.host,
                cli_args.port,
                cli_args.n_min.into(),
                cli_args.n_max.into(),
                cli_args.check_unused.try_into().unwrap(),
                cli_args.max_age.try_into().unwrap(),
            )
            .await;
        })
}