tempest 0.1.1

Realtime message handling framework inspired by Apache Storm and built with Actix
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
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159

use crate::service::config::{get_topology_config, TopologyConfig};
use structopt::StructOpt;

/// PackageOpt provides an interface for
/// parsing Topology Package command line arguments
#[derive(Debug, Clone, StructOpt)]
#[structopt(name = "Tempest Package", about = "Topology cli options")]
pub(crate) struct PackageOpt {
    #[structopt(short = "h", long = "host", default_value = "0.0.0.0")]
    /// Topology host
    pub host: String,

    #[structopt(short = "p", long = "port", default_value = "8765")]
    /// Topology port
    pub port: String,

    #[structopt(long = "agent_host", default_value = "0.0.0.0")]
    /// Agent host
    pub agent_host: String,

    #[structopt(long = "agent_port", default_value = "7654")]
    /// Agent port
    pub agent_port: String,

    #[structopt(short = "g", long = "graceful_shutdown", default_value = "30000")]
    /// Graceful shutdown milliseconds
    pub graceful_shutdown: u64,

    #[structopt(subcommand)]
    pub cmd: PackageCmd,
}

impl PackageOpt {
    /// Returns the `{host}:{port}`
    pub fn host_port(&self) -> String {
        format!("{}:{}", self.host, self.port)
    }

    /// Returns the sub-command config
    pub fn get_config(&self) -> Result<Option<TopologyConfig>, config::ConfigError> {
        let cfg = match &self.cmd {
            PackageCmd::Task(ref opt) => &opt.config,
            PackageCmd::Topology(ref opt) => &opt.config,
            PackageCmd::Standalone(ref opt) => &opt.config,
        };
        match &cfg {
            Some(ConfigOpt::Config { path }) => {
                if let Some(path) = path {
                    match get_topology_config(path) {
                        Ok(cfg) => return Ok(Some(cfg)),
                        Err(err) => return Err(err),
                    }
                }
            }
            _ => {}
        }
        Ok(None)
    }
}

/// Package sub-commands
#[derive(Debug, Clone, StructOpt)]
pub(crate) enum PackageCmd {
    /// Standalone option for running a topology
    /// and all tasks as a single, multi-threaded process
    #[structopt(name = "standalone")]
    Standalone(StandaloneOpt),

    /// Run a topology (source & pipeline) process
    #[structopt(name = "topology")]
    Topology(TopologyOpt),

    /// Run a task process by name
    #[structopt(name = "task")]
    Task(TaskOpt),
}

/// Sub-command for running a Standalone process
#[derive(Debug, Clone, StructOpt)]
pub(crate) struct StandaloneOpt {
    #[structopt(subcommand)]
    /// Topology.toml config option sub-command
    pub config: Option<ConfigOpt>,
}

/// Sub-command for running a Topology process
#[derive(Debug, Clone, StructOpt)]
pub(crate) struct TopologyOpt {
    #[structopt(subcommand)]
    /// Topology.toml config
    pub config: Option<ConfigOpt>,
}

/// Sub-command for running a Task process
#[derive(Default, Debug, Clone, StructOpt)]
pub(crate) struct TaskOpt {
    #[structopt(short = "n", long = "name")]
    /// Name of the topology task to run
    pub name: String,

    // Not currently implemented
    // #[structopt(short = "w", long = "workers")]
    // /// Number of workers to spin up for this task per node
    // pub workers: Option<u64>,
    #[structopt(short = "i", long = "poll_interval")]
    /// Poll interval milliseconds
    pub poll_interval: Option<u64>,

    #[structopt(short = "c", long = "poll_count")]
    /// Number of messages to read per poll
    pub poll_count: Option<u16>,

    #[structopt(short = "b", long = "max_backoff")]
    /// Max backoff wait in milliseconds
    pub max_backoff: Option<u64>,

    #[structopt(subcommand)]
    /// Topology.toml config
    pub config: Option<ConfigOpt>,
}

/// Config option
#[derive(Debug, Clone, StructOpt)]
pub(crate) enum ConfigOpt {
    #[structopt(name = "config")]
    /// Topology.toml config
    Config {
        #[structopt(short = "p", long = "path")]
        path: Option<String>,
        // TODO: implement toml string loading
        // #[structopt(short = "t", long = "toml")]
        // /// Parse this String as a toml table
        // toml: Option<String>,
    },
}

/// Command options for configuring an Agent.
///
#[derive(Default, Debug, Clone, StructOpt)]
pub(crate) struct AgentOpt {
    #[structopt(short = "h", long = "host", default_value = "0.0.0.0")]
    /// Agent host
    pub host: String,

    #[structopt(short = "p", long = "port", default_value = "7654")]
    /// Agent port
    pub port: String,
}

impl AgentOpt {
    pub fn new(host: String, port: String) -> Self {
        AgentOpt { host, port }
    }

    /// Returns the `{host}:{port}`
    pub fn host_port(&self) -> String {
        format!("{}:{}", self.host, self.port)
    }
}