dora-cli 1.0.0

`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
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

use std::{collections::BTreeMap, io::Write};

use clap::Args;
use dora_core::config::InputMapping;
use dora_message::id::{DataId, NodeId};
use serde::Serialize;
use tabwriter::TabWriter;

use crate::{
    command::{
        Executable, default_tracing,
        topic::selector::{DataflowSelector, node_topic_inputs, node_topic_outputs},
    },
    common::CoordinatorOptions,
    formatting::OutputFormat,
};

/// Print the outputs defined in a dataflow descriptor.
///
/// Examples:
///
/// List topics in a dataflow:
///   dora topic list -d my-dataflow
///
/// List topics as JSON:
///   dora topic list -d my-dataflow --format json
#[derive(Debug, Args)]
#[clap(verbatim_doc_comment)]
pub struct List {
    #[clap(flatten)]
    selector: DataflowSelector,

    /// Output format
    ///
    /// `json` emits JSON Lines (one object per line).
    #[clap(long, value_name = "FORMAT", default_value_t = OutputFormat::Table)]
    pub format: OutputFormat,

    #[clap(flatten)]
    coordinator: CoordinatorOptions,
}
impl Executable for List {
    fn execute(self) -> eyre::Result<()> {
        default_tracing()?;

        list(self.coordinator, self.selector, self.format)
    }
}

#[derive(Serialize)]
struct OutputEntry {
    node: NodeId,
    name: DataId,
    subscribers: Vec<String>,
}

fn list(
    coordinator: CoordinatorOptions,
    selector: DataflowSelector,
    format: OutputFormat,
) -> eyre::Result<()> {
    let session = coordinator.connect()?;
    let (_dataflow_id, descriptor) = selector.resolve(&session)?;

    let mut subscribers = BTreeMap::<(NodeId, DataId), Vec<(NodeId, DataId)>>::new();
    for node in &descriptor.nodes {
        for (input_id, input) in node_topic_inputs(node) {
            if let InputMapping::User(user) = &input.mapping {
                subscribers
                    .entry((user.source.clone(), user.output.clone()))
                    .or_default()
                    .push((node.id.clone(), input_id.clone()));
            }
        }
    }

    let mut entries = Vec::new();
    for node in &descriptor.nodes {
        for output in node_topic_outputs(node) {
            let subscribers_for_output = subscribers.remove(&(node.id.clone(), output.clone()));
            entries.push(OutputEntry {
                node: node.id.clone(),
                name: output,
                subscribers: subscribers_for_output
                    .unwrap_or_default()
                    .into_iter()
                    .map(|(node, data)| format!("{node}/{data}"))
                    .collect(),
            });
        }
    }

    match format {
        OutputFormat::Table => {
            let mut tw = TabWriter::new(std::io::stdout().lock());
            tw.write_all(b"Node\tName\tSubscribers\n")?;
            for entry in entries {
                tw.write_all(
                    format!(
                        "{}\t{}\t{}\n",
                        entry.node,
                        entry.name,
                        entry.subscribers.join(", ")
                    )
                    .as_bytes(),
                )?;
            }
            tw.flush()?;
        }
        OutputFormat::Json => {
            for entry in entries {
                println!("{}", serde_json::to_string(&entry)?);
            }
        }
    }

    Ok(())
}