Struct MqttTopic 

pub struct MqttTopic(/* private fields */);

A subscription descriptor for one MQTT topic.

A topic carries no wildcard, so it is a name a publisher can use: the subscription is reached again by publishing to it, share group or not, because the group takes that copy between its members. That is what lets this descriptor answer for its own retry copies, and what a subscription written with + or # cannot do - MqttFilter is the descriptor for those.

use std::time::Duration;

use ruststream_rumqttc::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Reading {
    device: String,
    temperature: f64,
}

#[subscriber(MqttTopic::new("devices/dev42/telemetry").shared("workers"))]
async fn store(reading: &Reading) -> HandlerOutcome {
    if reading.temperature.is_nan() {
        // A sensor glitch: ask for the reading again in five seconds.
        return HandlerOutcome::retry_after(Duration::from_secs(5));
    }
    tracing::info!(reading.device, reading.temperature, "reading stored");
    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| {
            // A topic addresses its own retry copies, so the mount site names no destination.
            b.include(store)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead/telemetry");
        },
    )
}

Implementations

impl MqttTopic

pub fn new(topic: impl Into<String>) -> Self

Names the topic. A value carrying a wildcard is refused when the subscription opens, naming MqttFilter as the descriptor that takes one.

pub fn qos(self, qos: Qos) -> Self

Sets the delivery quality of service. Defaults to Qos::AtLeastOnce.

pub fn shared(self, group: impl Into<String>) -> Self

Makes this an MQTT 5 shared subscription in group: the broker distributes matching messages across the group’s consumers instead of fanning out to each.

pub fn topic(&self) -> &str

The topic this subscribes to (without any share group).

Trait Implementations

impl Clone for MqttTopic

fn clone(&self) -> Self

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for MqttTopic

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.

impl Eq for MqttTopic

impl FromName for MqttTopic

A topic is all this descriptor needs to exist, so the mount site may supply it: a definition written #[subscriber(MqttTopic)] takes its topic from name(..) there, and the quality of service and share group stay the defaults.

fn from_name(name: impl Into<Cow<'static, str>>) -> Self

Builds the source bound to name.

impl PartialEq for MqttTopic

fn eq(&self, other: &Self) -> bool

Equality operator ==.

fn ne(&self, other: &Rhs) -> bool

Inequality operator !=.

impl RedeliveryAddressed<ConnectedMqttBroker> for MqttTopic

The plain topic, never the share group’s wire form: $share/<group>/<topic> is a subscribe-side name, and a publish to it would reach a topic of that literal spelling.

fn redelivery_address( &self, _connected: &ConnectedMqttBroker, ) -> impl Future<Output = Result<RedeliveryAddress, MqttError>> + Send

Where a publish reaches this subscription again.

impl StructuralPartialEq for MqttTopic

impl SubscriptionSource<ConnectedMqttBroker> for MqttTopic

type Copies = AddressedCopies

One topic is one destination, so the descriptor says where a copy of a delivery reaches this subscription again and the mount site names nothing.

async fn subscribe( self, connected: &ConnectedMqttBroker, ) -> Result<MqttSubscriber, MqttError>

The app’s subscription stays in the session when the service stops: its drop leaves the filter on the server.

fn operation_bindings(&self) -> Bindings

The quality of service this subscription receives at, which is the consumer’s own setting and belongs to the operation rather than the channel.

fn message_bindings(&self) -> Bindings

The MQTT 5 properties every delivery on this subscription is mapped through.

type Subscriber = MqttSubscriber

The subscriber type this source opens.

fn name(&self) -> &str

The name (subject / channel) this subscription binds to.

fn declare_retry(self, declaration: &RetryDeclaration) -> Self

where Self: Sized,

Takes the registration’s retry declaration into the descriptor, before the subscription opens.

fn declare_retry_on( &self, connected: &C, declaration: &RetryDeclaration, ) -> Result<(), DeclareRetryError>

Carries the registration’s declaration to the broker, at the same point and before the subscription opens.

fn channel_bindings(&self) -> Bindings

What this subscription adds to its channel in the generated AsyncAPI document.