Struct MqttFilter 

pub struct MqttFilter { /* private fields */ }

A subscription descriptor for one MQTT topic filter, wildcards included.

One such subscription reads every topic its filter matches, so it addresses none of them: the mount site names where a deferred retry copy goes. MqttTopic is the descriptor for a subscription that names a single topic and answers for its own copies.

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

#[derive(Deserialize)]
struct Telemetry {
    temperature: f64,
}

#[subscriber(MqttFilter::new("devices/+/telemetry").qos(Qos::AtLeastOnce).shared("workers"))]
async fn record(telemetry: &Telemetry, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    tracing::info!(%topic, telemetry.temperature, "reading recorded");
    HandlerOutcome::ack()
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("fleet", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "fleet-svc"),
        |b| {
            // The filter names no single topic, so the mount site says where a retry copy goes.
            b.include(record)
                .out_retry(Publish::default())
                .to("devices/retry/telemetry");
        },
    )
}

A publish transform names the destination per delivery instead of once. Naming a topic with .to(..) and composing a naming transform are mutually exclusive: writing both does not compile.

use ruststream::runtime::{Names, Outgoing, PublishContext};
use ruststream_rumqttc::prelude::*;
use serde::Deserialize;

#[derive(Deserialize)]
struct Command {
    action: String,
}

#[subscriber(MqttFilter::new("devices/+/commands").qos(Qos::AtLeastOnce))]
async fn execute(command: &Command, Ctx(topic): Ctx<DeliveryTopic>) -> HandlerOutcome {
    tracing::info!(command.action, %topic, "running");
    HandlerOutcome::ack()
}

/// Sends every copy back to the topic its own delivery arrived on, which is the one topic of
/// the filter's many that this message belongs to.
struct ToDeliveryTopic;

impl<Options> PublishTransform<ForReply<MqttContext>, Options> for ToDeliveryTopic {
    type Destination = Names;

    fn apply(
        &self,
        out: &mut Outgoing<'_>,
        _options: &mut Option<Options>,
        cx: &PublishContext<'_, MqttContext>,
    ) {
        out.set_name(cx.context(DeliveryTopic).to_owned());
    }
}

#[ruststream::app]
fn app() -> impl App {
    RustStream::new(AppInfo::new("commands", "0.1.0")).with_broker(
        MqttBroker::new("mqtt://localhost:1883", "commands-svc"),
        |b| {
            // A copy returns to the device it came from rather than to one topic for the fleet.
            b.include(execute)
                .max_attempts(nonzero!(5u32))
                .dead_letter("dead/commands")
                .out_retry(Publish::default())
                .transform(ToDeliveryTopic);
        },
    )
}

Implementations

impl MqttFilter

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

Names the topic filter, with wildcards as the protocol defines them.

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 filter(&self) -> &str

The plain topic filter (without any share group).

Trait Implementations

impl Clone for MqttFilter

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 MqttFilter

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

Formats the value using the given formatter.

impl Eq for MqttFilter

impl FromName for MqttFilter

A topic filter is all this descriptor needs to exist, so the mount site may supply it: a definition written #[subscriber(MqttFilter)] takes its filter 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 MqttFilter

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

Equality operator ==.

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

Inequality operator !=.

impl StructuralPartialEq for MqttFilter

impl SubscriptionSource<ConnectedMqttBroker> for MqttFilter

type Copies = NamedCopies

A filter matches many topics and is a topic itself only by accident, so the descriptor addresses no copy and the mount site names where one goes.

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.