MqttPublish

MqttPublish, Publish in the prelude, sets the QoS and the retain flag; a publish returns once the client session owns the message.

A PUBLISH packet carries two arguments besides the topic: a quality of service and the retain flag. The policy declares both: qos defaults to Qos::AtLeastOnce, and retain to false. A mount site names it wherever a registration publishes: .out_reply(Publish::default()) for a reply, .out(Slot, Publish::default()) for a slot, .out_retry(Publish::default()) for retry copies. A scope's after_startup hook takes one too, for a message the service sends once it is connected. The policy holds no connection; the runtime pairs it with the connected broker at startup into an MqttPublisher. A reply mounted without a policy sends through Publish::default(), to the topic its type declares. A publish returns once the client session owns the message, not once the server confirmed it. At QoS 1 and 2 the session sends the message again until the server acknowledges it, across reconnects. A topic with a + or # wildcard is refused with MqttError::Publish. One message changes the arguments with the qos and retain steps.

Example

A device agent answers each command and each ping on the topic the answer's type names:

Macros

examples/mqtt_publishing_publish.rs

#[subscriber(MqttTopic::new("devices/dev42/commands"), reply)]
async fn acknowledge(command: &Command) -> CommandAck {
    tracing::info!(command.action, "command applied");
    CommandAck {
        action: command.action.clone(),
    }
}

#[subscriber(MqttTopic::new("devices/dev42/ping"), reply)]
async fn pong(ping: &Ping) -> Pong {
    Pong { seq: ping.seq }
}

Manual

examples/manual/mqtt_publishing_publish.rs

struct Acknowledge;

impl Handle<Command, CommandAck> for Acknowledge {
    fn handle(
        &self,
        command: &Command,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<CommandAck, HandlerOutcome>> {
        tracing::info!(command.action, "command applied");
        ready(Ok(CommandAck {
            action: command.action.clone(),
        }))
    }
}

struct Answer;

impl Handle<Ping, Pong> for Answer {
    fn handle(
        &self,
        ping: &Ping,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Pong, HandlerOutcome>> {
        ready(Ok(Pong { seq: ping.seq }))
    }
}

The acknowledgement goes out exactly once; the pong takes the default policy:

Macros

examples/mqtt_publishing_publish.rs

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "device-agent-dev42"),
        |b| {
            // An acknowledgement must arrive exactly once.
            b.include(acknowledge)
                .out_reply(Publish::default().qos(Qos::ExactlyOnce));
            // No policy named: the reply goes through the broker's default one.
            b.include(pong);
        },
    )
}

Manual

examples/manual/mqtt_publishing_publish.rs

fn app() -> impl App {
    RustStream::new(AppInfo::new("device-agent", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "device-agent-dev42"),
        |b| {
            // An acknowledgement must arrive exactly once.
            b.include(
                subscriber("devices/dev42/commands", Acknowledge)
                    .reply()
                    .build(),
            )
            .out_reply(Publish::default().qos(Qos::ExactlyOnce));
            // No policy named: the reply goes through the broker's default one.
            b.include(subscriber("devices/dev42/ping", Answer).reply().build());
        },
    )
}