cufflink-cli 0.22.0

CLI for the Cufflink CRUD microservice platform — deploy, init, and manage services
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

//! `cufflink consumers ls`: the durables of the current service.

use super::{json_answer, listing};

pub async fn ls(env: Option<&str>) -> eyre::Result<()> {
    let (config, service_id, service_name) = super::deployments::resolve_service(env).await?;
    let client = config.http_client();
    let resp = config
        .auth_request(
            &client,
            reqwest::Method::GET,
            &format!("{}/api/services/{}/consumers", config.api_url, service_id),
        )
        .send()
        .await?;
    let body = json_answer(resp, "list consumers").await?;
    print_consumers(&service_name, &body);
    Ok(())
}

fn print_consumers(service_name: &str, body: &serde_json::Value) {
    let mut table = listing(&[
        "HANDLER",
        "SUBJECT",
        "STREAM",
        "DURABLE",
        "STATE",
        "PENDING",
        "IN_FLIGHT",
        "REDELIVERED",
        "HERE",
    ]);
    for consumer in body["consumers"].as_array().into_iter().flatten() {
        let server = &consumer["server"];
        let count = |key: &str| {
            server[key]
                .as_u64()
                .map_or_else(|| "-".to_string(), |n| n.to_string())
        };
        let state = match (server.is_null(), consumer["bound"].as_bool()) {
            (true, Some(true)) => "missing (bound)",
            (true, _) => "missing",
            (false, Some(true)) => "bound",
            (false, _) => "managed",
        };
        table.add_row(vec![
            consumer["handler"].as_str().unwrap_or("?").to_string(),
            consumer["subject"].as_str().unwrap_or("?").to_string(),
            consumer["stream"].as_str().unwrap_or("?").to_string(),
            consumer["durable"].as_str().unwrap_or("?").to_string(),
            state.to_string(),
            count("num_pending"),
            count("num_ack_pending"),
            count("num_redelivered"),
            if consumer["running_here"] == true {
                "✓"
            } else {
                ""
            }
            .to_string(),
        ]);
    }
    println!("{table}");
    let handovers = body["handovers"].as_array().cloned().unwrap_or_default();
    println!(
        "\n{} consumer(s) of '{}' v{}",
        body["consumers"].as_array().map_or(0, Vec::len),
        service_name,
        body["version"]
    );
    if !handovers.is_empty() {
        let mut table = listing(&["DEPLOY", "STREAM", "FROM", "TO", "MODE", "START", "AT"]);
        for handover in &handovers {
            table.add_row(vec![
                format!("v{}", handover["deploy_version"]),
                handover["stream"].as_str().unwrap_or("?").to_string(),
                handover["from_name"].as_str().unwrap_or("?").to_string(),
                handover["to_name"].as_str().unwrap_or("?").to_string(),
                handover["mode"].as_str().unwrap_or("?").to_string(),
                handover["start_sequence"]
                    .as_i64()
                    .map_or_else(|| "-".to_string(), |n| n.to_string()),
                handover["created_at"]
                    .as_str()
                    .map(|at| at.chars().take(19).collect())
                    .unwrap_or_default(),
            ]);
        }
        println!("\nHandovers:\n{table}");
    }
}