krabka_barrier/
dispatch.rs

1//! Runs one command line: builds the client, sends the subcommand's request,
2//! and hands the response to the reporter.
3//!
4//! This is the only module that talks to a broker, so every exit code the tool
5//! can return is decided here or in `report`. A transport failure is kept
6//! distinct from a refusal throughout, because a request that did not complete
7//! says nothing at all about whether the broker acted on it.
8
9use {
10    crate::{
11        EXIT_REFUSED, EXIT_UNREACHABLE,
12        cli::{Cli, Command},
13        report::{report_alter, report_describe, report_list, report_trigger, report_verify},
14        verify,
15    },
16    clap::Parser,
17    krabka_protocol::krabka::barrier as api,
18    krabka_units::{Time, convert::TimeExt as _},
19};
20
21krabka_macros::parsed_cli_entrypoint!({
22    /// Run the tool from an argv-style iterator, returning its exit code.
23    ///
24    /// `0` means the broker accepted the request. `1` means it refused one, and the
25    /// reason is on stderr. `2` means the broker could not be reached, so nothing
26    /// is known about the outcome. `3` means a cut does not match the log.
27    ///
28    /// # Panics
29    ///
30    /// Panics if `argv` does not parse, which for a caller passing a literal
31    /// argument list is a bug in that list rather than a runtime condition.
32});
33
34/// Run one parsed command.
35pub async fn run(cli: Cli) -> i32 {
36    let client = krabka_macros::connect_cli_client!(
37        &cli.bootstrap_server,
38        "krabka-barrier",
39        EXIT_UNREACHABLE
40    );
41    dispatch(&client, cli.command).await
42}
43
44/// Send one command's request and print its response.
45async fn dispatch(client: &krabka_client_core::Client, command: Command) -> i32 {
46    match command {
47        Command::Define {
48            group,
49            topics,
50            interval,
51            retained_cuts,
52        } => {
53            send_alter(client, api::AlterableBarrierGroup {
54                group,
55                topics,
56                // -1 is the sentinel for "no periodic injection".
57                interval_ms: interval.map_or(-1, Time::millis_i64),
58                retained_cuts,
59                delete: false,
60                ..api::AlterableBarrierGroup::default()
61            })
62            .await
63        }
64        Command::Delete { group } => {
65            send_alter(client, api::AlterableBarrierGroup {
66                group,
67                delete: true,
68                ..api::AlterableBarrierGroup::default()
69            })
70            .await
71        }
72        Command::Describe { groups } => {
73            let request = api::DescribeBarrierGroupsRequest {
74                groups,
75                ..api::DescribeBarrierGroupsRequest::default()
76            };
77            send_then(client, request, report_describe).await
78        }
79        Command::Trigger { group, timeout } => {
80            let request = api::TriggerBarrierRequest {
81                group,
82                // A non-positive timeout asks the broker for its own default.
83                timeout_ms: timeout
84                    .map(Time::millis_i64)
85                    .and_then(|ms| i32::try_from(ms).ok())
86                    .unwrap_or(0),
87                ..api::TriggerBarrierRequest::default()
88            };
89            send_then(client, request, report_trigger).await
90        }
91        Command::List {
92            group,
93            from_epoch,
94            max_results,
95        } => {
96            let request = api::ListBarrierCutsRequest {
97                group,
98                from_epoch,
99                max_results,
100                ..api::ListBarrierCutsRequest::default()
101            };
102            send_then(client, request, report_list).await
103        }
104        Command::Verify { group, epoch } => match verify::verify(client, &group, epoch).await {
105            Ok(outcome) => report_verify(&outcome),
106            Err(error) => {
107                eprintln!("{error}");
108                EXIT_REFUSED
109            }
110        },
111    }
112}
113
114async fn send_alter(client: &krabka_client_core::Client, group: api::AlterableBarrierGroup) -> i32 {
115    send_then(
116        client,
117        api::AlterBarrierGroupsRequest {
118            groups: vec![group],
119            ..api::AlterBarrierGroupsRequest::default()
120        },
121        report_alter,
122    )
123    .await
124}
125
126/// Send one request and hand its response to `report`.
127///
128/// A transport failure is printed as such and ends on `EXIT_UNREACHABLE`,
129/// because a request that did not complete says nothing about whether the
130/// broker acted on it.
131async fn send_then<R: krabka_client_core::ProtocolRequest>(
132    client: &krabka_client_core::Client,
133    request: R,
134    report: impl FnOnce(&R::Response) -> i32,
135) -> i32 {
136    match client.send(request).await {
137        Ok(response) => report(&response),
138        Err(error) => {
139            eprintln!("the request did not complete, so its outcome is unknown: {error}");
140            EXIT_UNREACHABLE
141        }
142    }
143}
krabka_barrier/dispatch.rs

krabka_barrier/
dispatch.rs
