tawasuyu/brahman

1//! `subscriber` — cliente brahman que loguea cada `MatchEvent` recibido.
2//!
3//! Declara una Card con un input `in` de tipo `json`. Cada vez que el
4//! broker matchea (o desmatch) ese input contra un productor, imprime
5//! una línea. Útil para visualizar la dinámica del broker en vivo.
6//!
7//! Uso:
8//! ```sh
9//! cargo run -p brahman-handshake --example subscriber [label]
10//! ```
11
12use std::collections::BTreeSet;
13use std::time::Duration;
14
15use brahman_card::{
16    ulid::Ulid, Card, Flow, Flows, Lifecycle, Payload, Priority, Supervision, TypeRef,
17    CARD_SCHEMA_VERSION,
18};
19use brahman_handshake::{client::Client, transport};
20
21#[tokio::main(flavor = "current_thread")]
22async fn main() -> anyhow::Result<()> {
23    let label = std::env::args()
24        .nth(1)
25        .unwrap_or_else(|| "subscriber".into());
26
27    let card = Card {
28        schema_version: CARD_SCHEMA_VERSION,
29        id: Ulid::new(),
30        label: label.clone(),
31        provides: BTreeSet::new(),
32        requires: BTreeSet::new(),
33        payload: Payload::Virtual,
34        supervision: Supervision::OneShot,
35        lifecycle: Lifecycle::Daemon,
36        priority: Priority::Normal,
37        flow: Flows {
38            input: vec![Flow {
39                name: "in".into(),
40                ty: TypeRef::Primitive {
41                    name: "json".into(),
42                },
43                pin_to: None,
44            }],
45            output: vec![],
46        },
47        ..Default::default()
48    };
49
50    let path = transport::default_socket_path();
51    eprintln!("[{label}] connecting to {}", path.display());
52    let mut client = Client::connect(&path, card).await?;
53    eprintln!(
54        "[{label}] attached: session={} init={}",
55        client.session(),
56        client.server_info().init_attached
57    );
58
59    // Loop: espera hasta 25s por un MatchEvent. Si timeout, ping para
60    // mantener la conexión viva.
61    loop {
62        match client.await_event(Duration::from_secs(25)).await? {
63            Some(ev) => {
64                eprintln!(
65                    "[{label}] {:?}  {}  ←  {}.{}  via={:?}{}",
66                    ev.kind,
67                    ev.consumer_flow,
68                    if ev.producer_label.is_empty() {
69                        "<none>"
70                    } else {
71                        &ev.producer_label
72                    },
73                    ev.producer_flow,
74                    ev.via,
75                    if ev.pinned { "  📌" } else { "" }
76                );
77            }
78            None => {
79                let _ts = client.ping().await?;
80            }
81        }
82    }
83}