tawasuyu/brahman

1//! `brahman-demo-consumer` — registra una Card con un `flow.input`
2//! que matchea al producer demo, se queda escuchando MatchEvents.
3//!
4//! Variables de entorno:
5//! - `BRAHMAN_DEMO_LABEL` — label de la Card. Default `demo-consumer`.
6//! - `BRAHMAN_DEMO_FLOW` — nombre del flow.input. Default `demo-feed`.
7//! - `BRAHMAN_DEMO_TYPE` — type primitive. Default `json` (debe matchear
8//!   con el producer para que veamos un Match en el broker).
9
10use std::collections::BTreeSet;
11use std::time::Duration;
12
13use brahman_card::{
14    Card, CgroupSpec, Flow, Flows, NamespaceSet, Payload, ResourceLimits, SomaSpec, Supervision,
15    TypeRef, CARD_SCHEMA_VERSION,
16};
17use brahman_handshake::client::Client;
18use brahman_handshake::transport;
19use ulid::Ulid;
20
21#[tokio::main(flavor = "current_thread")]
22async fn main() -> Result<(), Box<dyn std::error::Error>> {
23    tracing_subscriber::fmt()
24        .with_env_filter(
25            tracing_subscriber::EnvFilter::try_from_default_env()
26                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
27        )
28        .init();
29
30    let label =
31        std::env::var("BRAHMAN_DEMO_LABEL").unwrap_or_else(|_| "demo-consumer".to_string());
32    let flow_name = std::env::var("BRAHMAN_DEMO_FLOW").unwrap_or_else(|_| "demo-feed".into());
33    let type_name = std::env::var("BRAHMAN_DEMO_TYPE").unwrap_or_else(|_| "json".into());
34
35    let card = build_consumer_card(&label, &flow_name, &type_name);
36    let socket = transport::default_socket_path();
37
38    tracing::info!(
39        socket = %socket.display(),
40        label = %label,
41        flow = %flow_name,
42        ty = %type_name,
43        "demo-consumer conecta y queda escuchando MatchEvents"
44    );
45
46    let mut client = Client::connect(&socket, card).await?;
47    tracing::info!(session = %client.session(), "session asignada");
48
49    // Drenamos eventos cada 5s. Cada tick los logueamos si hay alguno;
50    // si no, ping de heartbeat (mismo razonamiento que producer).
51    loop {
52        let evt = client.await_event(Duration::from_secs(5)).await?;
53        match evt {
54            Some(ev) => tracing::info!(?ev, "MatchEvent recibido"),
55            None => {
56                if let Err(e) = client.ping().await {
57                    tracing::warn!(?e, "ping falló tras quiet period — saliendo");
58                    break;
59                }
60            }
61        }
62    }
63
64    Ok(())
65}
66
67fn build_consumer_card(label: &str, flow_name: &str, type_name: &str) -> Card {
68    Card {
69        schema_version: CARD_SCHEMA_VERSION,
70        id: Ulid::new(),
71        lineage: None,
72        label: label.into(),
73        provides: BTreeSet::new(),
74        requires: BTreeSet::new(),
75        soma: SomaSpec {
76            cgroup: CgroupSpec {
77                path: "ente.slice/demo".into(),
78                cpu_weight: None,
79                io_weight: None,
80            },
81            namespaces: NamespaceSet::default(),
82            rlimits: ResourceLimits::default(),
83            cpu_affinity: None,
84        },
85        payload: Payload::Virtual,
86        supervision: Supervision::OneShot,
87        flow: Flows {
88            input: vec![Flow {
89                name: flow_name.into(),
90                ty: TypeRef::Primitive {
91                    name: type_name.into(),
92                },
93                pin_to: None,
94            }],
95            output: vec![],
96        },
97        ..Default::default()
98    }
99}