wireband-edge 0.4.1

Lightweight Wire.Band client — semantic data middleware for any domain (IoT, AI/ML, DeFi, legal, geospatial, supply chain, and more)
Documentation 
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
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

//! wireband-edge CLI
//!
//! Run a Wire.Band edge gateway from the command line.
//!
//! # Examples
//!
//!     wireband-edge --broker mqtt://localhost:1883 --topics "sensors/#" "machines/#"
//!
//!     wireband-edge \
//!         --broker mqtts://broker.example.com:8883 \
//!         --topics "factory/line1/#" \
//!         --backend http://my-server:8000 \
//!         --device-id factory-rpi4 \
//!         --delta-threshold 0.02

use std::time::Duration;

use clap::Parser;
use tracing::info;
use tracing_subscriber::{EnvFilter, fmt};

use wireband_edge::client::{ClientConfig, WireBandClient};

// ---------------------------------------------------------------------------
// CLI args
// ---------------------------------------------------------------------------

#[derive(Parser, Debug)]
#[command(
    name = "wireband-edge",
    version,
    about = "Wire.Band edge gateway — bridge MQTT to a Wire.Band backend"
)]
struct Args {
    /// MQTT broker URL  (mqtt:// or mqtts://)
    #[arg(short, long, default_value = "mqtt://localhost:1883")]
    broker: String,

    /// MQTT topic filters
    #[arg(short, long, num_args = 1.., default_value = "#")]
    topics: Vec<String>,

    /// Wire.Band backend HTTP URL
    #[arg(long, default_value = "http://localhost:8000")]
    backend: String,

    /// Unique identifier for this gateway node
    #[arg(short, long, default_value = "edge-node")]
    device_id: String,

    /// Bearer token for backend authentication
    #[arg(long)]
    api_key: Option<String>,

    /// Suppress numeric readings with fractional change below this value
    /// (0.0 = forward all, 0.02 = suppress if <2% change)
    #[arg(long, default_value_t = 0.0)]
    delta_threshold: f64,

    /// Flush interval in seconds
    #[arg(long, default_value_t = 1.0)]
    flush_interval: f64,

    /// Max events per HTTP flush request
    #[arg(long, default_value_t = 200)]
    flush_batch: usize,

    /// Local ring buffer capacity in events
    #[arg(long, default_value_t = 50_000)]
    buffer_size: usize,

    /// Max retry attempts per flush before backing off
    #[arg(long, default_value_t = 3)]
    max_retries: u32,

    /// Log level
    #[arg(long, default_value = "info")]
    log_level: String,
}

// ---------------------------------------------------------------------------
// Entry point
// ---------------------------------------------------------------------------

#[tokio::main]
async fn main() -> anyhow::Result<()> {
    let args = Args::parse();

    // Tracing — WIREBAND_LOG env var overrides --log-level
    let filter = EnvFilter::try_from_env("WIREBAND_LOG")
        .unwrap_or_else(|_| EnvFilter::new(&args.log_level));
    fmt().with_env_filter(filter).init();

    let config = ClientConfig {
        backend_url:     args.backend.clone(),
        device_id:       args.device_id.clone(),
        api_key:         args.api_key.clone(),
        buffer_size:     args.buffer_size,
        flush_interval:  Duration::from_secs_f64(args.flush_interval),
        flush_batch:     args.flush_batch,
        max_retries:     args.max_retries,
        delta_threshold: args.delta_threshold,
        ..Default::default()
    };

    let client = WireBandClient::new(config);
    client.start();

    info!(
        device_id  = %args.device_id,
        broker     = %args.broker,
        topics     = ?args.topics,
        backend    = %args.backend,
        delta      = args.delta_threshold,
        "Wire.Band edge starting"
    );

    #[cfg(feature = "mqtt")]
    {
        use wireband_edge::mqtt::MqttConnector;

        let result = MqttConnector::new(&args.broker, args.delta_threshold)
            .run(client.clone(), args.topics)
            .await;

        client.stop().await;

        if let Err(e) = result {
            eprintln!("MQTT error: {e}");
            std::process::exit(1);
        }
    }

    #[cfg(not(feature = "mqtt"))]
    {
        eprintln!(
            "No connector enabled. Rebuild with a connector feature, e.g.:\n\
             cargo build --features mqtt"
        );
        client.stop().await;
        std::process::exit(1);
    }

    Ok(())
}