[FEAT] publish / subscribe for Linkable contracts, migrate the MQTT demos, document input conventions #391

Part of #388. The last step: it needs #389 (detachable links, Input / Output) and #390 (connect → ConnectorHandle, OutboundLink::on / InboundLink::on) to have landed. Closing this issue completes #388.

No spike was needed. Everything here is thin code on top of #389 and #390, and the #388 prototype already showed the compile errors this layer produces.

1. publish / subscribe in aimdb-data-contracts

Feature linkable, in aimdb-data-contracts/src/linkable.rs, next to the existing LinkableRegistrarExt.

let mqtt = builder.connect(MqttConnector::new(broker_url));
reg.input(subscribe(mqtt, "commands/temp/indoor").with_qos(1))
   .output(publish(mqtt, "sensors/temp/indoor").with_qos(1).with_retain(false));

• publish::<T: Linkable>(conn: ConnectorHandle, address: &str) builds OutboundLink::on(conn, address).with_link_codec(link_codecs::Default). subscribe is the same with InboundLink.
• The record's schema must still be reported. Today linked_to / linked_from call reg.schema(T::NAME, T::VERSION) before linking. publish / subscribe are values built before a registrar exists, so they need their own small Output<T> / Input<T> wrapper types whose wire() calls reg.schema(..) and then reg.output(link) / reg.input(link).
  • The wrappers also give a readable error. [FEAT] Detachable link builders and Input / Output traits in core #389 notes that a raw OutboundLinkBuilder<Temp, Detached> in an E0277 reads poorly.
  • Connector options must keep working on them. Either the wrappers deref/forward to the link, or the wrapper is generic over the link and the extension traits (MqttLinkExt, Ros2LinkExt, LinkCodecBuilderExt) are implemented for it as well. Pick whichever needs less code; the extension traits must not be duplicated per connector.
• T: Linkable is a bound on publish / subscribe themselves, so a contract with no wire format is E0277 at the call.
• Per-link codec: either publish(..).with_link_codec(Json) overrides the default (preferred, if the wrapper forwards LinkCodecBuilderExt), or add publish_with(conn, address, codec). Keep this in line with the existing linked_to_with.
• address accepts a bare address or a full URL with a matching scheme ([FEAT] AimDbBuilder::connect returns a ConnectorHandle; links through a handle need no scheme #390), so publish(mqtt, SensorKey::TempIndoor.link_address().unwrap()) works with existing #[link_address] keys.

2. Rebuild linked_to / linked_from on the same path

Decided in #388 (comment): keep the existing LinkableRegistrarExt::linked_to(url) / linked_from(url) and LinkCodecRegistrarExt::linked_to_with / linked_from_with. Re-implement them as one-line wrappers over the same code as publish / subscribe, using OutboundLink::new(url) / InboundLink::new(url) instead of a handle, so there is one implementation. Behaviour, including the schema() call, must not change. Whether to deprecate them is a separate decision, made after the demos have moved over.

3. Migrate the two MQTT demos

These are the acceptance test for the whole feature. Behaviour must not change.

• examples/tokio-mqtt-connector-demo/src/main.rs (215 lines). Today:

  • .with_connector(MqttConnector…)
  • one hand-written .link_to(..).with_qos(1).with_retain(false).with_serializer(..)
  • two reg.linked_to(..)
  • one .link_from(..).with_deserializer(..)
  • one reg.linked_from(..)
  • three nearly identical producer functions (indoor_temp_producer …)

  Contracts come from examples/mqtt-connector-demo-common (Temperature, TemperatureCommand, #[derive(Linkable)] behind its data-contracts feature). Keys come from keys.rs with #[link_address = "mqtt://…"]. Target:

  • let mqtt = builder.connect(..)
  • publish / subscribe everywhere
  • the simulated thermometers as one Input type that takes an owned channel (SimThermometers { indoor, outdoor, server_room }) and is generic over T: From<Reading>, moved into sim.rs

  The [FEAT] reg.input / reg.output with connection handles, so a wrong wiring is a compile error #388 prototype got main.rs down from 215 to 84 lines, plus 78 in sim.rs.

• examples/embassy-mqtt-connector-demo/src/main.rs (546 lines). Today:

  • .with_connector(mqtt) and .with_connector(SerialServer::new(..)), the remote-access server, which has no links. Keep it on with_connector, or use connect and ignore the handle.
  • three .link_to(..).with_serializer(|_, t| Ok(t.to_json_vec()))
  • .link_from(..) for commands

  Check whether the embassy build enables the common crate's data-contracts feature (it serializes by hand today). publish needs Linkable. Target, as in the prototype:

  • a board module (heap, clocks, Ethernet, UART) that hands out NucleoInputs { b1, .. }
  • an application module holding only records
  • user button B1 as a real PushButton input, built with T: From<ButtonEvent>

  Must still build with cargo build -p embassy-mqtt-connector-demo --target thumbv8m.main-none-eabihf (see make around line 682). The prototype reached the original firmware size. Flashing is optional; say in the PR whether it was done.

• Moving a channel into a second record must be a compile error (E0382). Show it in a compile_fail doctest or a short note in the PR, not in the demo itself.

4. Document the conventions

In docs/aimdb-usage-guide.md (its "Using Connectors" section) and in the rustdoc of Input / publish:

• Inputs yield an event; contracts are built with From<Event>. An input is generic over its record and bounded by T: From<Event>, so one input type serves every contract that can be built from its event (start trigger, door contact, counter pulse). A device that sends its own JSON is an input over a connection: it parses with the existing with_deserializer and maps with T::from.
• Device channels are owned, non-Clone values. A board hands out its inputs as struct fields (let NucleoInputs { b1 } = board::init(..)), and an input takes one by value (PushButton::on(b1)). Using a channel for a second record is then use of moved value. Pins never leave the board module. This is a pattern, not a type in core.
• Where From<Event> impls live (open question 3 of [FEAT] reg.input / reg.output with connection handles, so a wrong wiring is a compile error #388). The orphan rule allows impl From<E> for C only in the crate of E or of C. Convention: driver crates own their event types and ship their inputs with them, and contract crates implement From<Event>. When neither crate can, use a newtype or a .transform().
• What is checked where. Wrong pairings (From missing, contract not Linkable, wrong record type) are compile errors. A second channel use is a compile error through ownership. "One input per record" is not a compile error: a second .input() / .source() / .link_from() fails build(), as it does today.
• Update the with_connector docs and the Quick Start examples to show connect + publish / subscribe, and keep one example of the raw link_to path as the escape hatch.

Acceptance criteria

• publish / subscribe in aimdb-data-contracts (feature linkable): they report the schema, forward connector options, give E0277 for a type that isn't Linkable, and accept #[link_address] keys.
• linked_to / linked_from / *_with re-implemented on the same path, with unchanged behaviour (the existing tests in linkable.rs / link_codec.rs still pass).
• Tests: publish / subscribe register the same links and schema as linked_to / linked_from; options such as with_qos apply; a compile_fail doctest for a type that isn't Linkable.
• tokio-mqtt-connector-demo migrated and still works against a broker. (Migrated in refactor(examples): shared MQTT demo crate on publish / subscribe (#391, stage 2) #401; verified against the dev/mosquitto bench broker: all readings published, both commands received.)
• embassy-mqtt-connector-demo migrated, with a board module and a PushButton input on B1; builds for thumbv8m.main-none-eabihf.
• A compile_fail example of using a channel twice (E0382).
• Usage guide and rustdoc cover the conventions above; CHANGELOG entry.

Testing notes

The commands and caveats from #389's implementation notes apply here too: use the Makefile's feature sets, don't use --all-features on aimdb-core (defmt fails to link on the host), and run MQTT tests with --features std. For data-contracts, test both the full set (std,simulatable,migratable,observable,linkable-json,linkable-postcard,linkable-cdr,ros2) and --no-default-features --features alloc,linkable, so publish / subscribe also build without JSON.