ax 2.18.2

ax distributed event databank and command line tool
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

use crate::cmd::{AxCliCommand, ConsoleOpt};
use ax_core::{
    node_connection::{request_single, Task},
    util::formats::{
        events_protocol::{EventsRequest, EventsResponse},
        ActyxOSError, ActyxOSResult,
    },
};
use ax_sdk::types::service::OffsetsResponse;
use comfy_table::{presets::UTF8_FULL_CONDENSED, Cell, Table};
use futures::{stream, FutureExt, Stream};
use std::collections::BTreeSet;

#[derive(clap::Parser, Clone, Debug)]
/// obtain currently known offsets and replication targets
pub struct OffsetsOpts {
    #[command(flatten)]
    console_opt: ConsoleOpt,
}

pub struct EventsOffsets;
impl AxCliCommand for EventsOffsets {
    type Opt = OffsetsOpts;
    type Output = OffsetsResponse;

    fn run(opts: Self::Opt) -> Box<dyn Stream<Item = ActyxOSResult<Self::Output>> + Unpin> {
        Box::new(stream::once(
            async move {
                let (mut conn, peer) = opts.console_opt.connect().await?;
                request_single(
                    &mut conn,
                    move |tx| Task::Events(peer, EventsRequest::Offsets, tx),
                    |response| match response {
                        EventsResponse::Offsets(o) => Ok(o),
                        x => Err(ActyxOSError::internal(format!("Unexpected reply: {:?}", x))),
                    },
                )
                .await
            }
            .boxed(),
        ))
    }

    fn pretty(result: Self::Output) -> String {
        let OffsetsResponse { present, to_replicate } = result;
        let mut table = Table::new();
        table
            .load_preset(UTF8_FULL_CONDENSED)
            .set_header(["STREAM ID", "OFFSET", "TO REPLICATE"]);

        let streams = present
            .streams()
            .chain(to_replicate.keys().cloned())
            .collect::<BTreeSet<_>>();
        for s in streams.into_iter() {
            table.add_row([
                Cell::new(s),
                Cell::new(present.get(s).map(|x| x.to_string()).unwrap_or_default()),
                Cell::new(to_replicate.get(&s).map(|x| format!("+{}", *x)).unwrap_or_default()),
            ]);
        }
        table.to_string()
    }
}