feat(mqtt): encode embedded packets into one bbqueue write ring (design 054 §4.7)

Makefile

@@ -253,6 +253,8 @@ test:
 	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_broker
 	@printf "$(YELLOW)  → Testing MQTT connector (event-driven session: wake cadence, partial packets, QoS 1)$(NC)\n"
 	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test session_loop
+	@printf "$(YELLOW)  → Testing MQTT connector (embedded packet-size limits against a fake broker)$(NC)\n"
+	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test write_ring_proofs
 	@printf "$(YELLOW)  → Testing MQTT connector (the same criteria over mqtts://)$(NC)\n"
 	cargo test --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_session
 	@printf "$(YELLOW)  → Testing MQTT connector (no_std unit tests: framing, deadlines, TLS duplex)$(NC)\n"
@@ -402,6 +404,8 @@ clippy:
 	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_broker -- -D warnings
 	@printf "$(YELLOW)  → Clippy on MQTT connector (event-driven session criteria)$(NC)\n"
 	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test session_loop -- -D warnings
+	@printf "$(YELLOW)  → Clippy on MQTT connector (embedded packet-size limits)$(NC)\n"
+	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tokio-broker" --test write_ring_proofs -- -D warnings
 	@printf "$(YELLOW)  → Clippy on MQTT connector (the same criteria over mqtts://)$(NC)\n"
 	cargo clippy --package aimdb-mqtt-connector --no-default-features --features "_test-tls-broker" --test tls_session -- -D warnings
 	@printf "$(YELLOW)  → Clippy on MQTT connector (no_std unit tests)$(NC)\n"

aimdb-mqtt-connector/Cargo.toml

@@ -49,6 +49,8 @@ embedded = [
     # `mountain-mqtt`'s packet types take `heapless::Vec`, so the session names
     # it to build them. Same 0.8 the dependency itself resolves.
     "dep:heapless",
+    # The session's write ring.
+    "dep:bbqueue",
 ]
 
 # Convenience bundle: `embedded` plus the Embassy transport and clock. The
@@ -157,6 +159,9 @@ embassy-time = { version = "0.5.1", optional = true }
 embassy-sync = { version = "0.8.0", path = "../_external/embassy/embassy-sync", optional = true }
 embassy-futures = { workspace = true, optional = true }
 heapless = { workspace = true, optional = true }
+# Byte ring the embedded session encodes outbound packets into. `AtomicCoord`
+# needs atomic CAS, which every supported embedded target has.
+bbqueue = { version = "0.7", default-features = false, features = ["alloc"], optional = true }
 embassy-net = { version = "0.9.0", optional = true, features = [
     "tcp",
     "dhcpv4",

aimdb-mqtt-connector/src/embedded/mod.rs

@@ -13,6 +13,7 @@ pub(crate) mod session;
 // replace the polled loop.
 pub(crate) mod packet_reader;
 pub(crate) mod session_loop;
+pub(crate) mod write_ring;
 
 // TLS transport + SNTP time source.
 #[cfg(feature = "embassy-tls")]

aimdb-mqtt-connector/src/embedded/session.rs

@@ -64,13 +64,17 @@ where
     use mountain_mqtt::data::quality_of_service::QualityOfService;
 
     use crate::embedded::session_loop::run_session;
+    use crate::embedded::write_ring::{WriteRing, DEFAULT_WRITE_BUFFER};
 
     // Built once and borrowed for the loop; re-sent on every connection.
     let subscribe_topics: alloc::vec::Vec<(&str, QualityOfService)> = topics
         .iter()
         .map(|topic| (topic.as_str(), QualityOfService::Qos1))
         .collect();
 
+    // Allocated once and reused by every session.
+    let ring = WriteRing::new(DEFAULT_WRITE_BUFFER);
+
     loop {
         let mut stream = match dialer.connect(&host, port).await {
             Ok(stream) => stream,
@@ -92,6 +96,7 @@ where
             &subscribe_topics,
             &events,
             &actions,
+            &ring,
             &settings,
             &dialer,
             runtime.as_ref(),