MqttTopic

MqttTopic subscribes to one topic with no wildcard, so a retry copy published to that topic comes back with nothing named at the mount site.

MqttTopic::new takes the topic. qos sets the quality of service, AtLeastOnce by default. shared puts the subscription in a share group. The subscription's name is the topic: a test reads its deliveries under that name, and the topic is the channel of the AsyncAPI document. A topic is also a name a publisher can use. So a retry_after copy is published to the subscription's own topic, and the mount site names no destination for it. The copy is an ordinary publish, so every subscription whose filter matches the topic receives it too. A share group does not change where the copy goes: it is published to the plain topic. An empty topic, a topic with a + or # wildcard, or one with a control character stops the service at startup, before anything is subscribed. A wildcard belongs to MqttFilter, and the error names it.

Example

Commands for one valve, read at QoS 2; a command the valve refuses comes back five seconds later.

Macros
Manual

examples/mqtt_subscriptions_topic.rs
/// Runs the commands sent to dev42; the valve refuses the first attempt while it settles, so the
/// command comes back as a copy five seconds later.
#[subscriber(MqttTopic::new("devices/dev42/commands").qos(Qos::ExactlyOnce))]
async fn execute(command: &Command, ctx: &mut Context) -> HandlerOutcome {
    if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
        tracing::info!(command.id, "valve settling, trying again");
        return HandlerOutcome::retry_after(SETTLE);
    }
    tracing::info!(command.id, command.open, "command executed");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("valves", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "valves-svc"),
        |b| {
            // The topic is where a copy goes, so the mount site names nothing.
            b.include(execute);
        },
    )
}

examples/manual/mqtt_subscriptions_topic.rs
/// Runs the commands sent to dev42; the valve refuses the first attempt while it settles, so the
/// command comes back as a copy five seconds later.
struct Execute;

impl Handle<Command> for Execute {
    fn handle(
        &self,
        command: &Command,
        _outs: &(),
        ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        if ctx.headers().get_str(RETRY_COUNT_HEADER).is_none() {
            tracing::info!(command.id, "valve settling, trying again");
            return ready(Err(HandlerOutcome::retry_after(SETTLE)));
        }
        tracing::info!(command.id, command.open, "command executed");
        ready(Ok(()))
    }
}

fn app() -> impl App {
    RustStream::new(AppInfo::new("valves", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "valves-svc"),
        |b| {
            let commands = MqttTopic::new("devices/dev42/commands").qos(Qos::ExactlyOnce);
            // The topic is where a copy goes, so the mount site names nothing.
            b.include(
                subscriber("devices/dev42/commands", Execute)
                    .build()
                    .map_source(|_| commands),
            );
        },
    )
}

Example​

Example