dora-cli 0.2.0-rc-10

`dora` goal is to be a low latency, composable, and distributed data flow.
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

use crate::{
    check::{coordinator_running, daemon_running},
    control_connection,
};
use communication_layer_request_reply::TcpRequestReplyConnection;
use dora_core::topics::ControlRequest;
use eyre::Context;
use std::{fs, path::Path, process::Command, time::Duration};

#[derive(Debug, Default, serde::Serialize, serde::Deserialize)]
struct UpConfig {}

pub(crate) fn up(
    config_path: Option<&Path>,
    coordinator: Option<&Path>,
    daemon: Option<&Path>,
) -> eyre::Result<()> {
    let UpConfig {} = parse_dora_config(config_path)?;

    if !coordinator_running()? {
        start_coordinator(coordinator).wrap_err("failed to start dora-coordinator")?;
        // sleep a bit until the coordinator accepts connections
        while !coordinator_running()? {
            std::thread::sleep(Duration::from_millis(50));
        }
    }
    if !daemon_running()? {
        start_daemon(daemon).wrap_err("failed to start dora-daemon")?;
    }

    Ok(())
}

pub(crate) fn destroy(
    config_path: Option<&Path>,
    session: &mut Option<Box<TcpRequestReplyConnection>>,
) -> Result<(), eyre::ErrReport> {
    let UpConfig {} = parse_dora_config(config_path)?;

    if coordinator_running()? {
        // send destroy command to dora-coordinator
        control_connection(session)?
            .request(&serde_json::to_vec(&ControlRequest::Destroy).unwrap())
            .wrap_err("failed to send destroy message")?;
        println!("Send destroy command to dora-coordinator");
    } else {
        eprintln!("The dora-coordinator is not running");
    }

    Ok(())
}

fn parse_dora_config(config_path: Option<&Path>) -> Result<UpConfig, eyre::ErrReport> {
    let path = config_path.or_else(|| Some(Path::new("dora-config.yml")).filter(|p| p.exists()));
    let config = match path {
        Some(path) => {
            let raw = fs::read_to_string(path)
                .with_context(|| format!("failed to read `{}`", path.display()))?;
            serde_yaml::from_str(&raw)
                .with_context(|| format!("failed to parse `{}`", path.display()))?
        }
        None => Default::default(),
    };
    Ok(config)
}

fn start_coordinator(coordinator: Option<&Path>) -> eyre::Result<()> {
    let coordinator = coordinator.unwrap_or_else(|| Path::new("dora-coordinator"));

    let mut cmd = Command::new(coordinator);
    cmd.spawn()
        .wrap_err_with(|| format!("failed to run {}", coordinator.display()))?;

    println!("started dora coordinator");

    Ok(())
}

fn start_daemon(daemon: Option<&Path>) -> eyre::Result<()> {
    let daemon = daemon.unwrap_or_else(|| Path::new("dora-daemon"));

    let mut cmd = Command::new(daemon);
    cmd.spawn()
        .wrap_err_with(|| format!("failed to run {}", daemon.display()))?;

    println!("started dora daemon");

    Ok(())
}