diff --git a/src/config.rs b/src/config.rs index bd2fb81..955a189 100644 --- a/src/config.rs +++ b/src/config.rs @@ -10,6 +10,9 @@ pub struct Config { pub uplink_url: String, pub go2rtc_bin: String, pub go2rtc_config: String, + /// Path to the fleet config file - see `fleet.rs` and + /// `printers.example.toml` for the shape. + pub printers_path: String, /// Per-printer CA certificate overrides, keyed by printer ID. Most /// Bambu printers chain to the bundled shared CA (`certs/bambu_ca2.pem`) @@ -34,6 +37,7 @@ impl Config { uplink_url: env_or("CONTINUUM_UPLINK_URL", "wss://api.continuum.local/ws/edge/v1"), go2rtc_bin: env_or("CONTINUUM_GO2RTC_BIN", "go2rtc"), go2rtc_config: env_or("CONTINUUM_GO2RTC_CONFIG", "./go2rtc.yaml"), + printers_path: env_or("CONTINUUM_PRINTERS_PATH", "printers.toml"), bambu_ca_cert_overrides: parse_cert_overrides(&env_or("CONTINUUM_BAMBU_CA_CERT_OVERRIDES", "")), bambu_require_valid_cert: env_or("CONTINUUM_BAMBU_REQUIRE_VALID_CERT", "true") == "true", diff --git a/src/main.rs b/src/main.rs index e1cbc3e..8b2df10 100644 --- a/src/main.rs +++ b/src/main.rs @@ -1,4 +1,5 @@ mod config; +mod fleet; mod go2rtc; mod printer; mod uplink; @@ -6,6 +7,7 @@ mod uplink; use tracing_subscriber::EnvFilter; use crate::config::Config; +use crate::printer::{GenericPrinter, PrinterHandle}; #[tokio::main] async fn main() -> anyhow::Result<()> { @@ -18,12 +20,24 @@ async fn main() -> anyhow::Result<()> { tracing::info!(gateway_id = %config.gateway_id, farm_id = %config.farm_id, "starting continuum-proxy"); - // Two things run at once for now: the cloud uplink, and the go2rtc - // camera-restreaming process. More will join them later - printer - // discovery, the plate changer - as you build past src/printer/'s - // stubs. `tokio::select!` runs both and stops as soon as either one - // returns; both loop forever below, so in practice this runs until the - // process is killed. + // Loaded and connected here, once, at startup - nothing yet re-reads + // printers.toml on a running daemon (adding/removing a printer means + // a restart), and nothing yet lets the uplink dispatch a command to + // one of these once connected - see PrinterHandle's own doc comment + // for what's built and what's still a stub per vendor. Keeping + // `fleet` alive as a local variable for the rest of `main` (which + // never returns, below) is what keeps every connection open; it's + // not otherwise touched yet. + let mut fleet = fleet::load(std::path::Path::new(&config.printers_path))?; + connect_fleet(&mut fleet).await; + + // Three things run at once for now: the cloud uplink, the go2rtc + // camera-restreaming process, and (implicitly, via the connections + // opened above) every configured printer's live session. More will + // join them later - printer discovery, the plate changer - as you + // build past src/printer/'s stubs. `tokio::select!` runs both of the + // tasks below and stops as soon as either one returns; both loop + // forever, so in practice this runs until the process is killed. tokio::select! { _ = uplink::run(config.clone()) => {} _ = go2rtc::watchdog(config.go2rtc_bin.clone(), config.go2rtc_config.clone()) => {} @@ -31,3 +45,25 @@ async fn main() -> anyhow::Result<()> { Ok(()) } + +/// Connects every configured printer concurrently (not one at a time - +/// Bambu's auto-detect fallback alone can take several seconds per +/// printer if it has to time out on MQTT v5 before trying v3.1.1; +/// serially, a fleet of twenty would take twenty times as long to start). +/// Logs success/failure per printer instead of failing startup over one - +/// a farm shouldn't refuse to start because one printer is unplugged. +async fn connect_fleet(fleet: &mut [PrinterHandle]) { + let results = + futures_util::future::join_all(fleet.iter_mut().map(|printer| async { + let name = printer.display_name(); + (name, printer.connect().await) + })) + .await; + + for (name, result) in results { + match result { + Ok(()) => tracing::info!(printer = %name, "connected"), + Err(err) => tracing::warn!(printer = %name, %err, "failed to connect"), + } + } +} diff --git a/src/printer/bambu_mqtt.rs b/src/printer/bambu_mqtt.rs index a39fb7c..e92a217 100644 --- a/src/printer/bambu_mqtt.rs +++ b/src/printer/bambu_mqtt.rs @@ -262,24 +262,28 @@ async fn poll_loop_v2(mut event_loop: rumqttc::v5::EventLoop, pending: PendingTa } /// Parses an incoming report and, if its `sequence_id` matches something -/// `request` is waiting on, hands it the parsed JSON. A report with no -/// `sequence_id`, or one nothing's waiting on (an unsolicited status +/// `request` is waiting on, hands it the *unwrapped* body. A report with +/// no `sequence_id`, or one nothing's waiting on (an unsolicited status /// push, not a reply to anything), is silently ignored - most Bambu /// reports aren't replies to a specific request at all. +/// +/// A report is `{category: {"sequence_id": "...", ...}}` - one key, same +/// shape a request is published as. `request` publishes `body` and wants +/// `body`-shaped data back, not the wrapper around it, so this looks +/// inside whichever single category key is present (rather than +/// requiring the caller to already know it) and sends only that inner +/// value through - confirmed against a real printer's `get_version` +/// reply, which failed to parse as `GetVersionResponse` before this +/// unwrapped it (real device, real bug: `serde_json::from_value` on the +/// still-wrapped `{"info": {...}}` reported "missing field `module`", +/// since `module` lives one level deeper than that). async fn complete_pending(pending: &PendingTable, payload: &[u8]) { let Ok(value) = serde_json::from_slice::(payload) else { return }; - let Some(sequence_id) = find_sequence_id(&value) else { return }; + let Some(object) = value.as_object() else { return }; + let Some((_, body)) = object.iter().next() else { return }; + let Some(sequence_id) = body.get("sequence_id").and_then(Value::as_str) else { return }; - if let Some(sender) = pending.lock().await.remove(&sequence_id) { - let _ = sender.send(value); + if let Some(sender) = pending.lock().await.remove(sequence_id) { + let _ = sender.send(body.clone()); } } - -/// A report is `{category: {"sequence_id": "...", ...}}` - one key, same -/// shape as a request. Look inside whichever single category key is -/// present rather than requiring the caller to already know it. -fn find_sequence_id(report: &Value) -> Option { - let object = report.as_object()?; - let (_, category_body) = object.iter().next()?; - category_body.get("sequence_id")?.as_str().map(str::to_string) -}