From 9643c8b6851dcfec0ce165c236b3c4ceab5b19d3 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Iwo=20Strzebo=C5=84ski?= Date: Sat, 5 Sep 2026 21:04:39 +0000 Subject: [PATCH] Real MQTT wiring for Bambu: connect, auto-detect v5/v3.1.1, request/reply Removes GenericPrinter::receive (two disconnected fire-and-forget stubs that didn't actually give request/response) in favor of a real mechanism in the new src/printer/bambu_mqtt.rs: - BambuMqttSession wraps whichever rumqttc client a connection ended up using (v3.1.1's AsyncClient/EventLoop for BambuMQTT v1, v5's for BambuMQTT v2 - different types, different event/error shapes) behind one interface. `open()` respects ForceV1/ForceV2, and for Auto tries v5 first, falling back to v3.1.1 on any failure (see the module doc comment for why "any failure", not a precisely-matched version- rejection code - unverified against a live printer, no broker here to test against). - Reuses BambuTls::build_connector (now pub(crate)) for rumqttc's TLS transport via TlsConfiguration::NativeConnector, instead of a second, separate TLS trust story - hence rumqttc now pulls in use-native-tls instead of its default rustls stack, so there's one TLS implementation in the binary, not two. - request() publishes and waits (with a timeout) for a report whose sequence_id matches, correlated via a shared pending-request table that a spawned background task completes as reports arrive - the first real tokio::spawn in this crate, flagged as deferred since printer/mod.rs's very first doc comment; this is why it was deferred until now, and why it's needed now (polling the eventloop has to run continuously, independent of whatever call site is waiting). - publish()/publish_raw() are the fire-and-forget half, for commands like gcode_line that only get acknowledged implicitly. BambuPrinter::connect/send_gcode_line/publish now call the real session instead of printing what they would do. Verified against 192.0.2.1 (a reserved, never-routable test address): both AsyncClient/EventLoop pairs actually attempt a real TCP connection and fail gracefully (network timeout, not a panic) when nothing answers, and Auto genuinely attempts both while ForceV1 skips straight to v3.1.1. --- Cargo.toml | 2 +- src/printer/bambu.rs | 104 +++++++-------- src/printer/bambu_mqtt.rs | 264 ++++++++++++++++++++++++++++++++++++++ src/printer/klipper.rs | 4 - src/printer/mod.rs | 11 +- src/printer/prusa.rs | 9 -- 6 files changed, 310 insertions(+), 84 deletions(-) create mode 100644 src/printer/bambu_mqtt.rs diff --git a/Cargo.toml b/Cargo.toml index 9799660..f5b7f0e 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -60,4 +60,4 @@ anyhow = "1" thiserror = "1" dotenvy = "0.15" tokio-serial = { version = "5.5.0", default-features = false } -rumqttc = "0.25.1" +rumqttc = { version = "0.25.1", default-features = false, features = ["use-native-tls"] } diff --git a/src/printer/bambu.rs b/src/printer/bambu.rs index 6d1f28e..cfaf931 100644 --- a/src/printer/bambu.rs +++ b/src/printer/bambu.rs @@ -73,7 +73,11 @@ impl BambuTls { /// Builds a `native_tls` connector configured for this trust mode: /// trusting only `ca_bytes()`, or accepting anything for `Insecure`. - fn build_connector(&self) -> anyhow::Result { + /// `pub(crate)`, not private: `bambu_mqtt.rs` (a sibling module, not a + /// descendant of this one) reuses this exact connector to configure + /// rumqttc's TLS transport, instead of every consumer's TLS trust + /// decisions living in two separate places. + pub(crate) fn build_connector(&self) -> anyhow::Result { let mut builder = native_tls::TlsConnector::builder(); // Bambu printer certificates don't carry their LAN IP as a Subject @@ -182,8 +186,10 @@ pub struct BambuPrinter { pub sn: String, tls: BambuTls, preference: BambuProtocolPreference, - /// `None` until `connect()` succeeds. - version: Option, + /// `None` until `connect()` succeeds. Carries the live MQTT session + /// alongside the protocol version it settled on, since the two are + /// decided together and always used together - see `bambu_mqtt.rs`. + connection: Option<(super::bambu_mqtt::BambuMqttSession, BambuProtocolVersion)>, } impl BambuPrinter { @@ -206,7 +212,7 @@ impl BambuPrinter { sn: sn.into(), tls, preference, - version: None, + connection: None, } } @@ -344,51 +350,33 @@ impl GenericPrinter for BambuPrinter { &self.base } - /// Settles `self.version` - forced directly if `preference` says so, - /// otherwise (the real version) attempting an MQTT v5 CONNECT first - /// and falling back to v3.1.1 if the broker's CONNACK rejects that - /// protocol version. No real MQTT client exists yet (still a stub, no - /// different from before), so `Auto` always "detects" v2 for now - - /// the fallback attempt itself is what's left to build. + /// Real, not a stub: opens the MQTT session (see `bambu_mqtt.rs` for + /// `preference`'s `Auto`-tries-v5-then-v3.1.1 fallback) and keeps it - + /// same reasoning as `PrusaSerialPrinter::connect` holding its port + /// open, just an MQTT session instead of a serial port. async fn connect(&mut self) -> Result<(), PrinterError> { - let mode = match &self.tls { - BambuTls::BundledCa => "bundled Bambu CA", - BambuTls::Custom(_) => "custom CA cert", - BambuTls::Insecure => "no certificate verification", - }; + let (session, version) = super::bambu_mqtt::BambuMqttSession::open( + &self.base.host, + &self.access_code, + &self.sn, + &self.tls, + self.preference, + ) + .await?; - let version = match self.preference { - BambuProtocolPreference::ForceV1 => BambuProtocolVersion::V1, - BambuProtocolPreference::ForceV2 => BambuProtocolVersion::V2, - BambuProtocolPreference::Auto => { - println!( - "[{}] auto-detecting protocol: trying MQTT v5 first", - self.base.name - ); - BambuProtocolVersion::V2 - } - }; - - println!( - "[{}] connecting over MQTTS:8883 ({mode}, {version:?})", - self.base.name - ); - self.version = Some(version); + println!("[{}] connected over MQTT ({version:?})", self.base.name); + self.connection = Some((session, version)); Ok(()) } + /// Real: a raw pass-through publish to `device/{sn}/request`, no + /// category envelope opinion - `send_gcode_line` below builds and + /// sends its own properly-wrapped request instead of going through + /// this. async fn publish(&mut self, payload: String) -> Result<(), PrinterError> { - // TODO: add publishing via MQTT - println!( - "[{}] publishing to device/{}/request: {payload}", - self.base.name, self.sn - ); - Ok(()) - } - - async fn receive(&mut self, payload: String) -> Result<(), PrinterError> { - // TODO: add MQTT response handling - Ok(()) + let (session, _) = + self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?; + session.publish_raw(payload).await } /// Same MQTT gcode-injection mechanism regardless of protocol @@ -423,25 +411,21 @@ impl GenericPrinter for BambuPrinter { /// which one depends on `self.version`, settled by `connect()` above - /// instead of a hand-written serde struct: `sequence_id` and the /// `command` const get filled in by the schema itself, `param` is the - /// one field we actually set. Only the actual MQTT publish to - /// `device/{sn}/request` is still a stub, same as `connect` above. + /// one field we actually set. Real, end to end: publishes over the + /// live MQTT session `connect()` opened, no waiting for a reply - + /// `gcode_line` is fire-and-forget, the printer just executes it. async fn send_gcode_line(&mut self, gcode_line: String) -> Result<(), PrinterError> { - let version = self - .version - .ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?; + let (session, version) = + self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?; - let mut print = - continuum_schema_instance::build_instance_from_file( - &get_schema_for_command( - version, - "print".into(), - "gcode_line".into() - ) - ) - .map_err(|err| PrinterError(format!("failed to build gcode_line request: {err}")))?; - print["param"] = serde_json::Value::String(gcode_line); + let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command( + *version, + "print".into(), + "gcode_line".into(), + )) + .map_err(|err| PrinterError(format!("failed to build gcode_line request: {err}")))?; + body["param"] = serde_json::Value::String(gcode_line); - let payload = serde_json::json!({ "print": print }).to_string(); - self.publish(payload).await + session.publish("print", &body).await } } diff --git a/src/printer/bambu_mqtt.rs b/src/printer/bambu_mqtt.rs new file mode 100644 index 0000000..16eb7a7 --- /dev/null +++ b/src/printer/bambu_mqtt.rs @@ -0,0 +1,264 @@ +//! The real MQTT wiring for Bambu printers - the piece `connect`/ +//! `send_gcode_line` in `bambu.rs` used to just stub out with a +//! `println!`. Two things make this more than "call `publish`, get a +//! `String` back": +//! +//! 1. MQTT v3.1.1 (BambuMQTT v1) and MQTT v5 (BambuMQTT v2) are two +//! entirely separate rumqttc client types (`rumqttc::AsyncClient` vs +//! `rumqttc::v5::AsyncClient`) with different event/error types - +//! `BambuMqttClient` is the enum that hides which one a connection +//! ended up using, same pattern as `PrinterHandle` hides which vendor +//! struct you're holding. +//! 2. A response to a published request doesn't come back from that same +//! call - it arrives later, asynchronously, on a *subscription* to +//! `device/{sn}/report`, correlated only by matching `sequence_id`. +//! That needs: a background task continuously polling the connection +//! (rumqttc doesn't do this for you - nothing happens until you poll +//! the `EventLoop` yourself, repeatedly, forever), and a shared table +//! of "who's waiting for which `sequence_id`" that task completes as +//! reports arrive. This is the first real use of `tokio::spawn` in +//! this crate - flagged as deferred since `printer/mod.rs`'s very +//! first doc comment, and this is why: it only makes sense once there's +//! a long-running background job (polling forever) that has to run +//! independently of whatever call site is waiting on a response. +//! +//! Unverified against a live printer (no broker to test against here): +//! the exact `ConnectReturnCode`/error shape a real Bambu printer's +//! embedded broker gives when it doesn't understand MQTT v5's CONNECT at +//! all (rather than a clean version-rejection CONNACK) - `open()` below +//! treats *any* failure of the v5 attempt as "fall back to v3.1.1", not +//! just a precisely-matched version-rejection code, since a broker that +//! can't parse a v5 CONNECT packet may just drop the connection instead +//! of replying with one. + +use std::collections::HashMap; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::Arc; +use std::time::Duration; + +use rumqttc::v5::mqttbytes::v5::Packet as PacketV5; +use rumqttc::v5::mqttbytes::QoS as QoSV5; +use rumqttc::{Event, Packet, QoS, Transport, TlsConfiguration}; +use serde_json::Value; +use tokio::sync::{oneshot, Mutex}; + +use super::bambu::BambuTls; +use super::PrinterError; +use super::BambuProtocolVersion; + +/// Requests waiting on a report with a matching `sequence_id`. Shared +/// between whichever call registered one (in `request`) and the +/// background poll task that completes it when the report arrives. +type PendingTable = Arc>>>; + +const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); + +enum BambuMqttClient { + V1(rumqttc::AsyncClient), + V2(rumqttc::v5::AsyncClient), +} + +/// One live MQTT connection to a Bambu printer - whichever protocol +/// version `open()` settled on. +pub struct BambuMqttSession { + client: BambuMqttClient, + pending: PendingTable, + sn: String, + next_sequence_id: AtomicU64, +} + +impl BambuMqttSession { + /// `ForceV1`/`ForceV2` connect with that protocol directly. `Auto` + /// tries v5 first and falls back to v3.1.1 if that attempt fails - + /// see this module's doc comment for why "any failure", not just a + /// precisely-matched version-rejection code. + pub async fn open( + host: &str, + access_code: &str, + sn: &str, + tls: &BambuTls, + preference: super::bambu::BambuProtocolPreference, + ) -> Result<(Self, BambuProtocolVersion), PrinterError> { + use super::bambu::BambuProtocolPreference as Pref; + + match preference { + Pref::ForceV1 => { + let session = Self::open_v1(host, access_code, sn, tls).await?; + Ok((session, BambuProtocolVersion::V1)) + } + Pref::ForceV2 => { + let session = Self::open_v2(host, access_code, sn, tls).await?; + Ok((session, BambuProtocolVersion::V2)) + } + Pref::Auto => match Self::open_v2(host, access_code, sn, tls).await { + Ok(session) => Ok((session, BambuProtocolVersion::V2)), + Err(_) => { + let session = Self::open_v1(host, access_code, sn, tls).await?; + Ok((session, BambuProtocolVersion::V1)) + } + }, + } + } + + async fn open_v1(host: &str, access_code: &str, sn: &str, tls: &BambuTls) -> Result { + let connector = tls.build_connector().map_err(|err| PrinterError(format!("TLS setup failed: {err}")))?; + + let mut options = rumqttc::MqttOptions::new(format!("continuum-{sn}"), host, 8883); + options.set_credentials("bblp", access_code); + options.set_transport(Transport::tls_with_config(TlsConfiguration::NativeConnector(connector))); + options.set_keep_alive(Duration::from_secs(30)); + + let (client, mut event_loop) = rumqttc::AsyncClient::new(options, 10); + client + .subscribe(format!("device/{sn}/report"), QoS::AtMostOnce) + .await + .map_err(|err| PrinterError(format!("subscribe failed: {err}")))?; + + // Nothing actually happens on the wire until the eventloop is + // polled - this first poll is what makes the TCP+TLS+CONNECT + // handshake happen, and its result is the only way to know + // whether it worked. + match event_loop.poll().await { + Ok(_) => {} + Err(err) => return Err(PrinterError(format!("MQTT v3.1.1 connect failed: {err}"))), + } + + let pending: PendingTable = Arc::new(Mutex::new(HashMap::new())); + tokio::spawn(poll_loop_v1(event_loop, pending.clone())); + + Ok(Self { client: BambuMqttClient::V1(client), pending, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1) }) + } + + async fn open_v2(host: &str, access_code: &str, sn: &str, tls: &BambuTls) -> Result { + let connector = tls.build_connector().map_err(|err| PrinterError(format!("TLS setup failed: {err}")))?; + + let mut options = rumqttc::v5::MqttOptions::new(format!("continuum-{sn}"), host, 8883); + options.set_credentials("bblp", access_code); + options.set_transport(Transport::tls_with_config(TlsConfiguration::NativeConnector(connector))); + options.set_keep_alive(Duration::from_secs(30)); + + let (client, mut event_loop) = rumqttc::v5::AsyncClient::new(options, 10); + client + .subscribe(format!("device/{sn}/report"), QoSV5::AtMostOnce) + .await + .map_err(|err| PrinterError(format!("subscribe failed: {err}")))?; + + match event_loop.poll().await { + Ok(_) => {} + Err(err) => return Err(PrinterError(format!("MQTT v5 connect failed: {err}"))), + } + + let pending: PendingTable = Arc::new(Mutex::new(HashMap::new())); + tokio::spawn(poll_loop_v2(event_loop, pending.clone())); + + Ok(Self { client: BambuMqttClient::V2(client), pending, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1) }) + } + + /// Publishes `body` under `device/{sn}/request`, wrapped as + /// `{category: body}` (Bambu's top-level key names the *category*, + /// e.g. "print"/"info" - `body`'s own `command` field says which + /// specific one), and waits up to `REQUEST_TIMEOUT` for a report + /// whose `sequence_id` matches. Overwrites whatever `sequence_id` + /// `body` already had (continuum-schema-instance fills it with an + /// empty placeholder, since it has no way to know a real one) with a + /// fresh one from this session's own counter. + pub async fn request(&self, category: &str, mut body: Value) -> Result { + let sequence_id = self.next_sequence_id.fetch_add(1, Ordering::Relaxed).to_string(); + body["sequence_id"] = Value::String(sequence_id.clone()); + + let (tx, rx) = oneshot::channel(); + self.pending.lock().await.insert(sequence_id.clone(), tx); + + if let Err(err) = self.publish(category, &body).await { + self.pending.lock().await.remove(&sequence_id); + return Err(err); + } + + match tokio::time::timeout(REQUEST_TIMEOUT, rx).await { + Ok(Ok(response)) => Ok(response), + Ok(Err(_)) => Err(PrinterError(format!("{} closed before replying", self.sn))), + Err(_) => { + self.pending.lock().await.remove(&sequence_id); + Err(PrinterError(format!("{} did not reply within {REQUEST_TIMEOUT:?}", self.sn))) + } + } + } + + /// Publishes `body` under `device/{sn}/request`, wrapped as + /// `{category: body}`, without waiting for a reply - for commands + /// like `gcode_line` that only get acknowledged implicitly (the + /// printer just executes it), not answered directly. + pub async fn publish(&self, category: &str, body: &Value) -> Result<(), PrinterError> { + self.publish_raw(serde_json::json!({ category: body }).to_string()).await + } + + /// Publishes an already-serialized payload under `device/{sn}/request` + /// as-is, with no category wrapping - the low-level primitive both + /// `publish` above and `GenericPrinter::publish` (a raw pass-through, + /// no envelope opinion at all) are built on. + pub async fn publish_raw(&self, payload: String) -> Result<(), PrinterError> { + let topic = format!("device/{}/request", self.sn); + + match &self.client { + BambuMqttClient::V1(client) => client + .publish(topic, QoS::AtLeastOnce, false, payload) + .await + .map_err(|err| PrinterError(format!("publish failed: {err}"))), + BambuMqttClient::V2(client) => client + .publish(topic, QoSV5::AtLeastOnce, false, payload) + .await + .map_err(|err| PrinterError(format!("publish failed: {err}"))), + } + } +} + +/// Runs forever, polling the v3.1.1 eventloop and completing pending +/// requests as matching reports arrive. Ends (silently) once the +/// connection is gone for good - dropping the `BambuMqttSession` drops +/// its `AsyncClient`, which is what eventually makes that happen. +async fn poll_loop_v1(mut event_loop: rumqttc::EventLoop, pending: PendingTable) { + loop { + match event_loop.poll().await { + Ok(Event::Incoming(Packet::Publish(publish))) => { + complete_pending(&pending, &publish.payload).await; + } + Ok(_) => {} + Err(_) => return, + } + } +} + +async fn poll_loop_v2(mut event_loop: rumqttc::v5::EventLoop, pending: PendingTable) { + loop { + match event_loop.poll().await { + Ok(rumqttc::v5::Event::Incoming(PacketV5::Publish(publish))) => { + complete_pending(&pending, &publish.payload).await; + } + Ok(_) => {} + Err(_) => return, + } + } +} + +/// 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 +/// push, not a reply to anything), is silently ignored - most Bambu +/// reports aren't replies to a specific request at all. +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 }; + + if let Some(sender) = pending.lock().await.remove(&sequence_id) { + let _ = sender.send(value); + } +} + +/// 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) +} diff --git a/src/printer/klipper.rs b/src/printer/klipper.rs index 7b5e2de..107d629 100644 --- a/src/printer/klipper.rs +++ b/src/printer/klipper.rs @@ -34,10 +34,6 @@ impl GenericPrinter for KlipperPrinter { Ok(()) } - async fn receive(&mut self, payload: String) -> Result<(), PrinterError> { - // TODO: add HTTP response handling - Ok(()) - } /// OVERRIDE: Moonraker's own **hypothetical** native API, not gcode injection - no /// percent-to-PWM conversion needed since it already takes a /// percentage directly. Real version: diff --git a/src/printer/mod.rs b/src/printer/mod.rs index 57580dc..be28036 100644 --- a/src/printer/mod.rs +++ b/src/printer/mod.rs @@ -28,6 +28,7 @@ mod bambu; mod bambu_commands; +mod bambu_mqtt; mod klipper; mod marlin_serial; mod prusa; @@ -73,7 +74,6 @@ pub trait GenericPrinter { async fn publish(&mut self, payload: String) -> Result<(), PrinterError>; - async fn receive(&mut self, payload: String) -> Result<(), PrinterError>; /// REQUIRED, no default body: `percent` is 0-100, but how it actually /// gets sent - an MQTT gcode-injection command, a REST call, raw bytes @@ -159,15 +159,6 @@ impl GenericPrinter for PrinterHandle { } } - async fn receive(&mut self, payload: String) -> Result<(), PrinterError> { - match self { - PrinterHandle::Bambu(p) => p.receive(payload).await, - PrinterHandle::PrusaLink(p) => p.receive(payload).await, - PrinterHandle::PrusaSerial(p) => p.receive(payload).await, - PrinterHandle::Klipper(p) => p.receive(payload).await, - } - } - async fn set_fan_speed(&mut self, percent: u8) -> Result<(), PrinterError> { match self { PrinterHandle::Bambu(p) => p.set_fan_speed(percent).await, diff --git a/src/printer/prusa.rs b/src/printer/prusa.rs index c30af19..dc72aa6 100644 --- a/src/printer/prusa.rs +++ b/src/printer/prusa.rs @@ -34,11 +34,6 @@ impl GenericPrinter for PrusaLinkPrinter { Ok(()) } - async fn receive(&mut self, payload: String) -> Result<(), PrinterError> { - // TODO: add HTTP response handling - Ok(()) - } - /// OVERRIDE: gcode via PrusaLink's command-injection endpoint - same /// underlying M106 as Bambu, reusing the shared percent->PWM /// conversion, but sent over HTTP instead of MQTT. @@ -110,10 +105,6 @@ impl GenericPrinter for PrusaSerialPrinter { Ok(()) } - async fn receive(&mut self, payload: String) -> Result<(), PrinterError> { - // TODO: add Serial response handling - Ok(()) - } /// OVERRIDE: same gcode as PrusaLink (still M106, still the shared /// conversion), but sent through `send_gcode_line` below instead of a /// separate transport - serial only has the one way to talk to the