diff --git a/examples/bambu_commands_demo.rs b/examples/bambu_commands_demo.rs index 5c14e4a..412eb49 100644 --- a/examples/bambu_commands_demo.rs +++ b/examples/bambu_commands_demo.rs @@ -36,8 +36,14 @@ fn main() { // --- print report: arrives under "print", NOT "pushing" (pushall // triggers this stream, it isn't a direct reply to it) - includes the // full ams/xcam example, with one empty tray and one loaded tray, to - // exercise the untagged Empty/Loaded split. + // exercise the untagged Empty/Loaded split. sequence_id/command added + // by hand here (the original real capture this was built from was + // itself a truncated example that happened to omit them, along with + // dozens of other real fields the schema defines - not proof the real + // message lacks them, see this struct's own doc comment). let real_print_report = r#"{ + "sequence_id": "1", + "command": "push_status", "ams": {"ams": [{"humidity":"4","id":"0","temp":"22.7","tray":[ {"id":"0"}, {"bed_temp":"0","bed_temp_type":"0","cols":["000000FF"],"drying_temp":"0","drying_time":"0","id":"1","nozzle_temp_max":"240","nozzle_temp_min":"190","remain":0,"tag_uid":"0000000000000000","tray_color":"000000FF","tray_diameter":"0.00","tray_id_name":"","tray_info_idx":"GFA00","tray_sub_brands":"","tray_type":"PLA","tray_uuid":"00000000000000000000000000000000","tray_weight":"0","xcam_info":"000000000000000000000000"} diff --git a/src/printer/bambu.rs b/src/printer/bambu.rs index 950fca7..d49df78 100644 --- a/src/printer/bambu.rs +++ b/src/printer/bambu.rs @@ -402,6 +402,57 @@ impl BambuPrinter { serde_json::from_value(response) .map_err(|err| PrinterError(format!("failed to parse get_version response: {err}"))) } + + /// Bambu-only, same reasoning as `get_version` above. Publishes + /// `pushing.pushall`, which - unlike `get_version` - doesn't reply + /// under its own category: it triggers/refreshes the ongoing + /// `print.push_status` stream instead. Fire-and-forget, same as + /// `send_gcode_line`: there's nothing to wait for here, since the + /// forced full update this triggers self-describes as such via its + /// own `msg` field once it arrives (see `PrintStatusCache::apply`), + /// whenever that turns out to be - waiting on it added two real bugs + /// (a race, and matching the wrong "print" message) for no benefit + /// once that was understood. + /// + /// Call this periodically for v1 printers specifically - v2 has no + /// `pushing` category at all (checked: `schemas/bambulab/v2/ + /// request.schema.json` only wires camera/info/print/system/upgrade/ + /// xcam). Whether v2 needs some *other* trigger to get + /// `print.push_status` flowing at all is still an open question (a + /// live test meant to check that lost its connection entirely, an + /// unrelated failure, not real evidence either way) - if it does, + /// that's a different method to add once identified, not a variant + /// of this one. A v2 connection makes this call a no-op rather than + /// an error, since `pushing.pushall` specifically is confirmed not to + /// exist for it regardless. + pub async fn pushall(&self) -> Result<(), PrinterError> { + let (session, version) = + self.connection.as_ref().ok_or_else(|| PrinterError(format!("{} is not connected", self.base.name)))?; + + if matches!(version, BambuProtocolVersion::V2) { + return Ok(()); + } + + let mut body = continuum_schema_instance::build_instance_from_file(&get_schema_for_command( + *version, + "pushing".into(), + "pushall".into(), + )) + .map_err(|err| PrinterError(format!("failed to build pushall request: {err}")))?; + body["push_target"] = serde_json::Value::from(1); + + session.publish("pushing", &body).await + } + + /// The current in-memory `print.push_status` object - `None` until at + /// least one has arrived (which, on v1, requires having called + /// `pushall` at least once since connecting; v2 pushes on its own). + pub async fn print_status(&self) -> Option { + match self.connection.as_ref() { + Some((session, _)) => session.print_status().await, + None => None, + } + } } impl GenericPrinter for BambuPrinter { diff --git a/src/printer/bambu_commands/print.rs b/src/printer/bambu_commands/print.rs index 1b17017..c7e559d 100644 --- a/src/printer/bambu_commands/print.rs +++ b/src/printer/bambu_commands/print.rs @@ -1,11 +1,29 @@ -//! The ongoing print-state push (root key "print"). Doesn't carry a -//! `CommandEnvelope` - no `sequence_id`/`command` seen on this one in the -//! real capture, and it may genuinely not have them (an async push, not a -//! reply to a specific request) - revisit if a real capture proves -//! otherwise. +//! The ongoing print-state push (root key "print"). The one real capture +//! this struct was originally built from didn't show `sequence_id`/ +//! `command` - turned out to be because that capture was itself a +//! truncated example (it's also missing dozens of other fields the real +//! schema defines), not because the message genuinely lacks them: +//! `push_status.schema.json` composes the shared envelope schema like +//! every other Bambu message, so it does carry `sequence_id` - which +//! matters a lot, see `BambuMqttSession`'s doc comment on how it's used +//! both to correlate a `pushall` reply and to detect a gap in the +//! ongoing organic stream. +use super::CommandEnvelope; -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] pub struct PrintReport { + #[serde(flatten)] + pub envelope: CommandEnvelope, + /// BambuMQTT v1 only: `1` on an automatic/organic push (a partial + /// update - only the fields that changed), `0` on a forced one (a + /// `pushing.pushall`-triggered reply - the full object). This is the + /// actual, reliable signal for which kind a given message is - + /// confirmed against a real printer, unlike trying to infer it from + /// request/reply timing (see `BambuMqttSession`'s `PrintStatusCache`, + /// which used to try that and got it wrong twice). Not in + /// `push_status.schema.json` yet - add it there too when you're back + /// in continuum-schemas. + pub msg: Option, pub ams: Option, pub ams_exist_bits: Option, pub insert_flag: Option, @@ -16,19 +34,57 @@ pub struct PrintReport { pub xcam_status: Option, // The real payload has a lot more fields than shown here (this // example was truncated) - add them the same way, one field per line, - // `Option` for anything that isn't always present. + // `Option` for anything that isn't always present. Remember to + // also add each new field to `merge_from` below. +} + +impl PrintReport { + /// Merges `newer`'s fields into `self`, in place - BambuMQTT v1's + /// organic `push_status` pushes only carry the fields that changed, + /// so any field `newer` doesn't have (`None`) leaves `self`'s + /// existing value alone instead of clobbering it. Not used for v2 + /// (always a full object) or for a `pushall`-triggered reply on + /// either version (also guaranteed full) - see `BambuMqttSession`. + pub fn merge_from(&mut self, newer: PrintReport) { + self.envelope = newer.envelope; + self.msg = newer.msg; + if newer.ams.is_some() { + self.ams = newer.ams; + } + if newer.ams_exist_bits.is_some() { + self.ams_exist_bits = newer.ams_exist_bits; + } + if newer.insert_flag.is_some() { + self.insert_flag = newer.insert_flag; + } + if newer.power_on_flag.is_some() { + self.power_on_flag = newer.power_on_flag; + } + if newer.tray_exist_bits.is_some() { + self.tray_exist_bits = newer.tray_exist_bits; + } + if newer.tray_is_bbl_bits.is_some() { + self.tray_is_bbl_bits = newer.tray_is_bbl_bits; + } + if newer.xcam.is_some() { + self.xcam = newer.xcam; + } + if newer.xcam_status.is_some() { + self.xcam_status = newer.xcam_status; + } + } } /// Bambu's own naming has the outer container and the inner array both /// called `ams` - `print.ams.ams`, not a typo here, just what the printer /// actually sends. -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] pub struct AmsState { pub ams: Vec, } /// One physical AMS unit (Bambu's numbering starts at 0 via `id`). -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] pub struct AmsUnit { pub humidity: String, pub id: String, @@ -44,20 +100,20 @@ pub struct AmsUnit { /// every field `Empty` allows, plus more) would incorrectly match `Empty` /// first, since a plain struct doesn't reject extra fields by default. /// Verified this the hard way - see git history. -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] #[serde(untagged)] pub enum AmsTray { Empty(EmptyTray), Loaded(LoadedTray), } -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] #[serde(deny_unknown_fields)] pub struct EmptyTray { pub id: String, } -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] pub struct LoadedTray { pub id: String, pub bed_temp: String, @@ -82,7 +138,7 @@ pub struct LoadedTray { /// Camera-based AI print monitoring settings (spaghetti detection, /// first-layer inspection, etc.). -#[derive(serde::Deserialize, Debug)] +#[derive(serde::Deserialize, Debug, Clone)] pub struct XcamSettings { pub allow_skip_parts: bool, pub buildplate_marker_detector: bool, diff --git a/src/printer/bambu_mqtt.rs b/src/printer/bambu_mqtt.rs index e92a217..10e27ff 100644 --- a/src/printer/bambu_mqtt.rs +++ b/src/printer/bambu_mqtt.rs @@ -1,6 +1,6 @@ //! 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 +//! `println!`. Three 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 @@ -11,21 +11,39 @@ //! 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. +//! `device/{sn}/report`. 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) - 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 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 +//! Two genuinely different correlation problems live here, confirmed +//! against real printers, not assumed: +//! - `info.get_version`'s reply *does* echo the `sequence_id` you sent +//! - `request`/`pending` (keyed by sequence_id) is correct for this, +//! and for anything else empirically confirmed to behave the same +//! way. +//! - `print.push_status` does NOT - the printer stamps it with its own +//! free-running counter regardless of what triggered it (confirmed +//! live: a `pushing.pushall` request's sequence_id never came back on +//! the reply). So there's no ID to match a `pushall` reply against at +//! all - `next_category`, keyed by category name instead of an ID, +//! is "wait for the *next* message under this category, whichever +//! one it is" - the only thing that actually works here. +//! 3. `print.push_status` also needs to be kept as running in-memory +//! state, not just handed to whoever happens to be waiting at the +//! moment it arrives - see `PrintStatusCache` below for why (short +//! version: v1 only ever sends partial updates, so something has to +//! remember what the fields *were* to produce a coherent whole +//! object). +//! +//! Unverified against a live printer (no way to force it in a test): 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 @@ -43,26 +61,114 @@ use serde_json::Value; use tokio::sync::{oneshot, Mutex}; use super::bambu::BambuTls; -use super::PrinterError; +use super::bambu_commands::PrintReport; use super::BambuProtocolVersion; +use super::PrinterError; -/// 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. +/// Requests waiting on a report with a matching `sequence_id` - only +/// valid for a command empirically confirmed to echo it back, like +/// `get_version`. Shared between whichever call registered one (in +/// `request`) and the background poll task that completes it. type PendingTable = Arc>>>; +/// Callers waiting on the *next* `(category, command)` message, +/// regardless of its `sequence_id` - for a reply with no usable +/// correlation ID at all, like `pushall`'s. Keyed by the pair, not just +/// category: a category like "print" carries many different commands +/// (`gcode_line`, `pause`, `push_status`, ...), and matching on category +/// alone means some unrelated "print" message arriving first would +/// silently satisfy the wait instead of the one actually meant - caught +/// live, see `request_next_of_category`'s doc comment. One slot per +/// pair: a second wait registered before the first resolves silently +/// replaces it (the first caller's `rx` then just sees the channel +/// close) - fine for now since nothing calls `pushall` concurrently with +/// itself. +type NextCategoryTable = Arc>>>; + const REQUEST_TIMEOUT: Duration = Duration::from_secs(10); +/// Old firmware has been observed taking noticeably longer than +/// `REQUEST_TIMEOUT` to answer `pushall` - confirmed live (a real P1P's +/// first reply arrived after that 10s window had already elapsed). +const NEXT_CATEGORY_TIMEOUT: Duration = Duration::from_secs(30); enum BambuMqttClient { V1(rumqttc::AsyncClient), V2(rumqttc::v5::AsyncClient), } +/// The running in-memory `print.push_status` object, plus enough state to +/// know how to fold the next update into it: +/// +/// - BambuMQTT v2 sends the *entire* object every time - always replace. +/// - BambuMQTT v1 only sends the fields that changed since last time - +/// merge into the existing object, or the picture goes incomplete the +/// instant anything not present in the latest push gets treated as +/// "unset". The exception is a `pushing.pushall`-triggered push, which +/// is always the full object - and v1 self-describes which kind a given +/// message is via its own `msg` field (`1` automatic/partial, `0` +/// forced/full), confirmed against a real printer. Two earlier attempts +/// at inferring this from request/reply *timing* instead (matching +/// `sequence_id`, then "wait for the next print/push_status message") +/// both turned out wrong in practice - the printer stamps `sequence_id` +/// from its own counter regardless of what triggered a push, and +/// category-only matching couldn't tell one "print" message from +/// another. `msg` needs none of that: it's on the message itself, +/// correct regardless of which client (or nothing at all) triggered it, +/// and regardless of timing. +/// - `last_sequence_id` exists to notice when v1's organic stream skipped +/// an update (network hiccup, a message this client wasn't fast enough +/// to read): if the incoming `sequence_id` isn't exactly one more than +/// the last one seen, the merged object is missing whatever changed in +/// the gap. There's no way to know what that baseline number even is +/// ahead of time - the printer runs its own counter, not one this +/// client started - so the very first `push_status` received after +/// connecting just seeds `last_sequence_id` rather than being checked +/// against anything. +struct PrintStatusCache { + report: Option, + last_sequence_id: Option, +} + +impl PrintStatusCache { + fn new() -> Self { + Self { report: None, last_sequence_id: None } + } + + fn apply(&mut self, incoming: PrintReport, version: BambuProtocolVersion) { + let incoming_seq = incoming.envelope.sequence_id.parse::().ok(); + // A forced/full update isn't missing anything by definition, so a + // sequence gap right after one (or on v2, where every message is + // full) isn't a real problem worth warning about. + let is_full_object = matches!(version, BambuProtocolVersion::V2) || incoming.msg != Some(1); + + if let (Some(last), Some(incoming_seq), false) = (self.last_sequence_id, incoming_seq, is_full_object) { + if incoming_seq != last.wrapping_add(1) { + eprintln!( + "print.push_status sequence gap: expected {}, got {incoming_seq} - the in-memory \ + object may be missing whatever changed in between", + last.wrapping_add(1) + ); + } + } + if let Some(incoming_seq) = incoming_seq { + self.last_sequence_id = Some(incoming_seq); + } + + match (&mut self.report, is_full_object) { + (_, true) => self.report = Some(incoming), + (Some(existing), false) => existing.merge_from(incoming), + (None, false) => self.report = Some(incoming), + } + } +} + /// One live MQTT connection to a Bambu printer - whichever protocol /// version `open()` settled on. pub struct BambuMqttSession { client: BambuMqttClient, pending: PendingTable, + next_category: NextCategoryTable, + print_status: Arc>, sn: String, next_sequence_id: AtomicU64, } @@ -124,9 +230,26 @@ impl BambuMqttSession { } let pending: PendingTable = Arc::new(Mutex::new(HashMap::new())); - tokio::spawn(poll_loop_v1(event_loop, pending.clone())); + let next_category: NextCategoryTable = Arc::new(Mutex::new(HashMap::new())); + let print_status = Arc::new(Mutex::new(PrintStatusCache::new())); + let sn_owned = sn.to_string(); + tokio::spawn(poll_loop_v1( + event_loop, + pending.clone(), + next_category.clone(), + print_status.clone(), + BambuProtocolVersion::V1, + sn_owned, + )); - Ok(Self { client: BambuMqttClient::V1(client), pending, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1) }) + Ok(Self { + client: BambuMqttClient::V1(client), + pending, + next_category, + print_status, + sn: sn.to_string(), + next_sequence_id: AtomicU64::new(1), + }) } async fn open_v2(host: &str, access_code: &str, sn: &str, tls: &BambuTls) -> Result { @@ -149,9 +272,26 @@ impl BambuMqttSession { } let pending: PendingTable = Arc::new(Mutex::new(HashMap::new())); - tokio::spawn(poll_loop_v2(event_loop, pending.clone())); + let next_category: NextCategoryTable = Arc::new(Mutex::new(HashMap::new())); + let print_status = Arc::new(Mutex::new(PrintStatusCache::new())); + let sn_owned = sn.to_string(); + tokio::spawn(poll_loop_v2( + event_loop, + pending.clone(), + next_category.clone(), + print_status.clone(), + BambuProtocolVersion::V2, + sn_owned, + )); - Ok(Self { client: BambuMqttClient::V2(client), pending, sn: sn.to_string(), next_sequence_id: AtomicU64::new(1) }) + Ok(Self { + client: BambuMqttClient::V2(client), + pending, + next_category, + print_status, + sn: sn.to_string(), + next_sequence_id: AtomicU64::new(1), + }) } /// Publishes `body` under `device/{sn}/request`, wrapped as @@ -163,26 +303,15 @@ impl BambuMqttSession { /// empty placeholder, since it has no way to know a real one) with a /// fresh one from this session's own counter. /// - /// Only correct for commands that genuinely get *one direct, - /// sequence_id-correlated reply* - `info.get_version` is the - /// canonical example (the envelope schema says as much: "Sequence ID - /// matching the original request"). It is NOT a general "send a - /// command, get the response" primitive - Bambu's own mapping isn't - /// 1:1: - /// - `pushing.pushall` doesn't reply at all - it triggers an *ongoing* - /// `print.push_status` stream, unrelated to this request's - /// `sequence_id`. Use `publish` (fire-and-forget) for it, the same - /// way `send_gcode_line` already does for `gcode_line`. - /// - Even where a field genuinely exists, which report carries it can - /// differ *by protocol version*: `print.calibration` data is bundled - /// inside `print.push_status` on BambuMQTT v1, but v2 gives it its - /// own dedicated report. Calling `request` for something like that - /// would just hang until `REQUEST_TIMEOUT` waiting for a reply that - /// was never going to arrive correlated by `sequence_id` - not a - /// crash, but a silent design bug if used on the wrong kind of - /// command. An ongoing-stream subscription (register interest in a - /// *category*, get every report under it as it arrives, not a single - /// `oneshot`) is a separate primitive this doesn't provide yet. + /// Only use this for a command *empirically confirmed* to echo the + /// sequence_id you send back on its reply - `info.get_version` is + /// verified to do this against real hardware. Don't assume every + /// command does just because the field exists on every message (they + /// all compose the same envelope schema): `print.push_status` also + /// has a `sequence_id`, but a live P1P proved the printer stamps it + /// with its own free-running counter regardless of what triggered + /// it - `pushall` uses `next_category` below instead, precisely + /// because this method's assumption doesn't hold for it. 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()); @@ -205,6 +334,51 @@ impl BambuMqttSession { } } + /// Publishes `body` under `device/{sn}/request`, wrapped as + /// `{category: body}`, then waits up to `NEXT_CATEGORY_TIMEOUT` for + /// the *next* `reply_category`/`reply_command` report to arrive - not + /// matched by any ID, just whichever one shows up next. For a command + /// whose reply carries no usable correlation ID at all (see + /// `request`'s doc comment) - currently just `pushall`, but written + /// generically since nothing about the mechanism is push_status- + /// specific. + /// + /// Matching needs *both* `reply_category` and `reply_command`, not + /// category alone - caught live: matching on category only let an + /// unrelated "print" message (that category carries many commands - + /// `gcode_line`, `pause`, `push_status`, ...) satisfy a `pushall` wait + /// before the real `push_status` reply had even arrived, so + /// `pushall()` returned `Ok` having consumed the wrong message + /// entirely, and the in-memory cache never actually got its update. + pub async fn request_next_of_category( + &self, + category: &str, + body: Value, + reply_category: &str, + reply_command: &str, + ) -> Result { + let key = (reply_category.to_string(), reply_command.to_string()); + let (tx, rx) = oneshot::channel(); + self.next_category.lock().await.insert(key.clone(), tx); + + if let Err(err) = self.publish(category, &body).await { + self.next_category.lock().await.remove(&key); + return Err(err); + } + + match tokio::time::timeout(NEXT_CATEGORY_TIMEOUT, rx).await { + Ok(Ok(response)) => Ok(response), + Ok(Err(_)) => Err(PrinterError(format!("{} closed before replying", self.sn))), + Err(_) => { + self.next_category.lock().await.remove(&key); + Err(PrinterError(format!( + "{} sent no {reply_category}/{reply_command} within {NEXT_CATEGORY_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 @@ -231,59 +405,138 @@ impl BambuMqttSession { .map_err(|err| PrinterError(format!("publish failed: {err}"))), } } + + /// The current in-memory `print.push_status` object, if at least one + /// has arrived since connecting - `None` until then. Cloned out from + /// behind the lock rather than handing out a guard, so callers aren't + /// holding the poll loop's own lock while they use it. + pub async fn print_status(&self) -> Option { + self.print_status.lock().await.report.clone() + } } -/// 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) { +/// Runs forever, polling the v3.1.1 eventloop and routing incoming +/// reports (pending-request correlation, `print.push_status` caching). +/// Exits - loudly, not silently like before (that silence is exactly +/// what made a real dropped connection to a live H2C look identical to +/// "the printer just isn't sending anything") - once the connection is +/// gone for good. Dropping the `BambuMqttSession` drops its `AsyncClient`, +/// which is what eventually makes that happen during normal shutdown; +/// this also fires on an unexpected disconnect, which is the case this +/// logging is actually for. No reconnect logic yet - that's the next +/// thing to add here, not something to fake by pretending the loop +/// didn't end. +async fn poll_loop_v1( + mut event_loop: rumqttc::EventLoop, + pending: PendingTable, + next_category: NextCategoryTable, + print_status: Arc>, + version: BambuProtocolVersion, + sn: String, +) { loop { match event_loop.poll().await { Ok(Event::Incoming(Packet::Publish(publish))) => { - complete_pending(&pending, &publish.payload).await; + handle_incoming_report(&pending, &next_category, &print_status, version, &publish.payload).await; } Ok(_) => {} - Err(_) => return, + Err(err) => { + eprintln!("[{sn}] MQTT v3.1.1 connection lost, no longer receiving reports: {err}"); + return; + } } } } -async fn poll_loop_v2(mut event_loop: rumqttc::v5::EventLoop, pending: PendingTable) { +async fn poll_loop_v2( + mut event_loop: rumqttc::v5::EventLoop, + pending: PendingTable, + next_category: NextCategoryTable, + print_status: Arc>, + version: BambuProtocolVersion, + sn: String, +) { loop { match event_loop.poll().await { Ok(rumqttc::v5::Event::Incoming(PacketV5::Publish(publish))) => { - complete_pending(&pending, &publish.payload).await; + handle_incoming_report(&pending, &next_category, &print_status, version, &publish.payload).await; } Ok(_) => {} - Err(_) => return, + Err(err) => { + eprintln!("[{sn}] MQTT v5 connection lost, no longer receiving reports: {err}"); + return; + } } } } -/// Parses an incoming report and, if its `sequence_id` matches something -/// `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. +/// Parses an incoming report, unwraps its single category key (a report +/// is `{category: {"sequence_id": "...", ...}}`, same shape a request is +/// published as - both `request` and `request_next_of_category` publish +/// `body` and want `body`-shaped data back, not the wrapper around it), +/// and: /// -/// 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]) { +/// 1. If its `sequence_id` matches something `request` is waiting on, +/// hands it the unwrapped body. +/// 2. If its `(category, command)` matches something +/// `request_next_of_category` is waiting on, hands it the unwrapped +/// body too - independent of 1, since the whole reason this table +/// exists is replies with no usable correlation ID. +/// 3. If the unwrapped body is a `print.push_status`, folds it into the +/// running `PrintStatusCache` (which decides full-vs-merge on its own, +/// from the message's own `msg` field - not from whether step 1 or 2 +/// matched, see `PrintStatusCache::apply`'s doc comment for why not) +/// - and *before* step 4 wakes anyone up, not after (see 4). +/// 4. Only now - once the cache is guaranteed up to date - actually sends +/// on whichever oneshot(s) matched in steps 1/2, waking `request`/ +/// `request_next_of_category`. +/// +/// The order of 3 and 4 is load-bearing, caught live: sending on the +/// oneshot first let a waiter's task resume and check `print_status()` +/// before this function had gotten around to updating it - `pushall()` +/// returned `Ok`, and the very next `print_status()` call still saw +/// `None`. Nothing here awaits between the cache write and the sends +/// below, so that race is now closed. +/// +/// A report matching neither table isn't an error - most Bambu reports +/// (including every organic `push_status` push) aren't a reply to +/// anything specific. +async fn handle_incoming_report( + pending: &PendingTable, + next_category: &NextCategoryTable, + print_status: &Arc>, + version: BambuProtocolVersion, + payload: &[u8], +) { let Ok(value) = serde_json::from_slice::(payload) 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 }; + let Some((category, body)) = object.iter().next() else { return }; - if let Some(sender) = pending.lock().await.remove(sequence_id) { + let command = body.get("command").and_then(Value::as_str); + + let request_sender = + match body.get("sequence_id").and_then(Value::as_str) { + Some(sequence_id) => pending.lock().await.remove(sequence_id), + None => None, + }; + let next_category_sender = match command { + Some(command) => next_category.lock().await.remove(&(category.clone(), command.to_string())), + None => None, + }; + + // push_status doesn't need to know whether this client solicited it - + // full-vs-partial is self-described by the message's own `msg` field, + // see `PrintStatusCache::apply`. + if category == "print" && command == Some("push_status") { + if let Ok(report) = serde_json::from_value::(body.clone()) { + print_status.lock().await.apply(report, version); + } + } + + if let Some(sender) = request_sender { + let _ = sender.send(body.clone()); + } + if let Some(sender) = next_category_sender { let _ = sender.send(body.clone()); } }