feat(052): runtime-neutral MQTT connector

[lxsaah]

Wave C of design 052. MQTT is the exception among the connectors and stays that way deliberately — see below — but the embedded backend now takes a caller-supplied transport, which is what makes a FreeRTOS port possible without touching this crate.

Why MQTT keeps two protocol clients

Unifying on mountain-mqtt was considered and rejected: it does not support QoS 2 (Qos2NotSupported; the codec has pubrec/pubrel/pubcomp, the state machine does not use them) and its tokio module has no TLS — while the Tokio connector exposes both. rumqttc cannot ride the neutral layer either: its Transport is a closed enum, so no stream can be injected.

So one MqttConnector<B> over two backends, which is the seam that keeps unification a later, small change — if QoS 2 and a TLS connection land in the fork, deleting Native is a deletion, not a rewrite.

Backend	Client	QoS	TLS
Native	rumqttc (std)	0–2	rustls
Embedded<D>	mountain-mqtt (no_std)	0–1	embedded-tls

Breaking

The Tokio path is unchanged — MqttConnector::new(&url) still. Embassy callers supply the transport:

// mqtt:// — the adapter owns the socket
MqttConnector::new(&url).transport(EmbassyNet::tcp(*stack, MQTT_RX.init(..), MQTT_TX.init(..)))

// mqtts:// — TLS keeps the stack: it resolves DNS itself and owns buffers across sessions
MqttConnector::new(&url).tls(stack, TlsOptions::new(..))

The two new statics are not new RAM: the previous commit already allocated them inside the connector as private StaticCells. They are now visible and sizeable by the caller.

On FreeRTOS the same line is .transport(LwipNet::tcp()), with no edit here.

Worth a look in review

run_with_subscriptions is gone. It binds embassy_net::Stack and cannot take a transport, so injecting one meant giving it up — and with it, reconnect-and-resubscribe. That behaviour is now explicit in transport::run_sessions, one loop for both plain and TLS, extracted from the TLS path that was already running it in production.

The transport seam is mountain-mqtt's own Connection, not core's ByteStream. The client needs receive_if_ready — a non-blocking peek — which ByteStream does not express and a TLS session cannot provide (this crate's own embassy_tls documents why). SocketTransport<D> bridges from core's StreamDialer for adapters whose stream also offers Read + Write + ReadReady, which is the path a new runtime takes.

EmbassyTcpStream gained embedded_io_async::{Read, Write, ReadReady} by delegation — design 052 §6 specified this and wave A had not delivered it.

tests/embassy_broker.rs drives the whole connector against a fake broker over two crossover-wired embassy-net stacks, asserting CONNECT and SUBSCRIBE reach the wire. The subscribe is the property run_with_subscriptions used to own; without it inbound routing dies silently on the first reconnect. CI has no mosquitto, hence the fake.

Not addressed

mqtts:// stays Embassy-only — set_unix_time has no neutral home (design §9). SNTP is likewise unported, since it serves only that path.

Verification

make check clean: 2 023 tests, 0 failures, all stages.