Prelude

One glob, ruststream_rumqttc::prelude, brings the core prelude, the descriptors, Qos, the policy as Publish and the context key.

use ruststream_rumqttc::prelude::*; is the one import a routes file on MQTT writes. It carries the core's own prelude, so #[subscriber], RustStream, HandlerOutcome and Out come with it. It adds MqttBroker, the descriptors MqttTopic and MqttFilter, their quality of service Qos, and the context type MqttContext with its key DeliveryTopic.

The publish policy comes under the name every broker crate uses: Publish is MqttPublish. So a mount site reads the same on every broker. Two broker preludes in one file collide on that name; the prefixed MqttPublish at the crate root tells them apart.

A handler body bounds its slots with capability traits, such as Out<impl Publisher>, and names no broker type. So a file of handlers imports ruststream::prelude::* alone. The one exception is a body that sets the QoS or the retain flag of one message: it needs MqttPublishSteps in scope and the slot bounded Out<impl Publisher<Options = MqttPublishOptions>, Marker>, both from this prelude. MqttError is named where it is handled, from the crate root.

Example

The descriptor, its QoS, the context key and the policy all come from the one glob.

Macros
Manual

examples/mqtt_broker_prelude.rs
use ruststream_rumqttc::prelude::*;

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce))]
async fn check(
    telemetry: &Telemetry,
    Ctx(topic): Ctx<DeliveryTopic>,
    Out(alerts): Out<impl Publisher>,
) -> HandlerOutcome {
    if telemetry.temperature <= 30.0 {
        return HandlerOutcome::ack();
    }
    if alerts.message(&Alert { topic }).publish().await.is_err() {
        return HandlerOutcome::retry();
    }
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            b.include(check)
                .out(DefaultSlot, Publish::default().qos(Qos::ExactlyOnce))
                .out_retry(Publish::default())
                .to("devices/retry/telemetry")
                .build();
        },
    )
}

examples/manual/mqtt_broker_prelude.rs
use ruststream_rumqttc::prelude::*;

struct Check;

impl<Entry> Handle<Telemetry, (), Outs<(Entry,)>, MqttContext> for Check
where
    Entry: OutEntry<DefaultSlot, Wire: Publisher>,
{
    async fn handle(
        &self,
        telemetry: &Telemetry,
        outs: &Outs<(Entry,)>,
        ctx: &mut Context<'_, MqttContext>,
    ) -> Result<(), HandlerOutcome> {
        if telemetry.temperature <= 30.0 {
            return Ok(());
        }
        let alert = Alert {
            topic: ctx.context(DeliveryTopic).to_owned(),
        };
        if outs
            .get(DefaultSlot)
            .message(&alert)
            .publish()
            .await
            .is_err()
        {
            return Err(HandlerOutcome::retry());
        }
        Ok(())
    }
}

fn telemetry() -> MqttFilter {
    MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce)
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
        |b| {
            let check = subscriber("devices/+/telemetry", Check).map_source(|_| telemetry());
            b.include(check.build())
                .out(DefaultSlot, Publish::default().qos(Qos::ExactlyOnce))
                .out_retry(Publish::default())
                .to("devices/retry/telemetry")
                .build();
        },
    )
}

Example​

Example