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

use crate::graph;
use dora_core::{config::OperatorId, descriptor::SINGLE_OPERATOR_DEFAULT_ID};
use eyre::{eyre, Context};
use std::{path::Path, process::Command};

pub fn build(dataflow: &Path) -> eyre::Result<()> {
    let descriptor = graph::read_descriptor(dataflow)?;
    let dataflow_absolute = if dataflow.is_relative() {
        std::env::current_dir().unwrap().join(dataflow)
    } else {
        dataflow.to_owned()
    };
    let working_dir = dataflow_absolute.parent().unwrap();

    let default_op_id = OperatorId::from(SINGLE_OPERATOR_DEFAULT_ID.to_string());

    for node in &descriptor.nodes {
        match &node.kind {
            dora_core::descriptor::NodeKind::Runtime(runtime_node) => {
                for operator in &runtime_node.operators {
                    run_build_command(operator.config.build.as_deref(), working_dir).with_context(
                        || {
                            format!(
                                "build command failed for operator `{}/{}`",
                                node.id, operator.id
                            )
                        },
                    )?;
                }
            }
            dora_core::descriptor::NodeKind::Custom(custom_node) => {
                run_build_command(custom_node.build.as_deref(), working_dir).with_context(|| {
                    format!("build command failed for custom node `{}`", node.id)
                })?
            }
            dora_core::descriptor::NodeKind::Operator(operator) => {
                run_build_command(operator.config.build.as_deref(), working_dir).with_context(
                    || {
                        format!(
                            "build command failed for operator `{}/{}`",
                            node.id,
                            operator.id.as_ref().unwrap_or(&default_op_id)
                        )
                    },
                )?
            }
        }
    }

    Ok(())
}

fn run_build_command(build: Option<&str>, working_dir: &Path) -> eyre::Result<()> {
    if let Some(build) = build {
        let mut split = build.split_whitespace();
        let mut cmd = Command::new(
            split
                .next()
                .ok_or_else(|| eyre!("build command is empty"))?,
        );
        cmd.args(split);
        cmd.current_dir(working_dir);
        let exit_status = cmd
            .status()
            .wrap_err_with(|| format!("failed to run `{}`", build))?;
        if exit_status.success() {
            Ok(())
        } else {
            Err(eyre!("build command returned an error code"))
        }
    } else {
        Ok(())
    }
}