ruststream_rumqttc/
context.rs

1//! [`MqttContext`]: what a delivery carries beyond its payload and headers.
2//!
3//! MQTT publishes to a topic and subscribes with a filter, so the two are not the same string
4//! whenever a filter carries a wildcard. The subscription name a handler reads with `ctx.name()`
5//! is the filter; the topic the message was actually published to is here, under the
6//! [`DeliveryTopic`] key.
7
8use ruststream::{BuildContext, ContextField, Field};
9
10use crate::message::MqttMessage;
11
12/// The per-delivery context of this broker.
13///
14/// Read a field with a key rather than by field access: `ctx.context(DeliveryTopic)` in a handler
15/// body, `cx.context(DeliveryTopic)` in a publish transform on the reply or the retry position.
16///
17/// # Examples
18///
19/// ```
20/// # mod demo {
21/// use ruststream_rumqttc::prelude::*;
22/// use serde::Deserialize;
23///
24/// #[derive(Deserialize)]
25/// struct Telemetry {
26///     temperature: f64,
27/// }
28///
29/// #[subscriber(MqttFilter::new("devices/+/telemetry"))]
30/// async fn record(telemetry: &Telemetry, ctx: &mut Context<'_, MqttContext>) -> HandlerOutcome {
31///     let device = ctx.context(DeliveryTopic).split('/').nth(1).unwrap_or_default();
32///     tracing::info!(device, telemetry.temperature, "reading recorded");
33///     HandlerOutcome::ack()
34/// }
35///
36/// #[ruststream::app]
37/// fn app() -> impl App {
38///     RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
39///         MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
40///         |b| {
41///             b.include(record)
42///                 .out_retry(Publish::default())
43///                 .to("devices/retry/telemetry");
44///         },
45///     )
46/// }
47/// # }
48/// # fn main() {}
49/// ```
50#[derive(Debug, Clone, PartialEq, Eq)]
51#[non_exhaustive]
52pub struct MqttContext {
53    topic: String,
54}
55
56impl MqttContext {
57    /// Builds the context of a delivery that arrived on `topic`.
58    #[must_use]
59    pub fn new(topic: impl Into<String>) -> Self {
60        Self {
61            topic: topic.into(),
62        }
63    }
64
65    /// The concrete topic this delivery was published to, never the filter that matched it.
66    #[must_use]
67    pub fn topic(&self) -> &str {
68        &self.topic
69    }
70}
71
72/// The key for [`MqttContext::topic`].
73///
74/// It reads the topic borrowed through `ctx.context(DeliveryTopic)`, and owned through the
75/// `Ctx<DeliveryTopic>` extractor, which is also what fixes a handler's context type to
76/// [`MqttContext`] without a `Context` parameter.
77///
78/// # Examples
79///
80/// ```
81/// # #[cfg(feature = "testing")]
82/// # mod demo {
83/// use std::error::Error;
84///
85/// use ruststream::testing::TestApp;
86/// use ruststream_rumqttc::prelude::*;
87/// use serde::{Deserialize, Serialize};
88///
89/// #[derive(Deserialize, Serialize, Outgoing)]
90/// struct Telemetry {
91///     temperature: f64,
92/// }
93///
94/// #[derive(Debug, PartialEq, Deserialize, Serialize, Outgoing)]
95/// #[outgoing(name = "alerts")]
96/// struct Alert {
97///     topic: String,
98/// }
99///
100/// #[subscriber(MqttFilter::new("devices/+/telemetry"))]
101/// async fn watch(
102///     telemetry: &Telemetry,
103///     Ctx(topic): Ctx<DeliveryTopic>,
104///     Out(alerts): Out<impl Publisher>,
105/// ) -> HandlerOutcome {
106///     if telemetry.temperature <= 30.0 {
107///         return HandlerOutcome::ack();
108///     }
109///     if alerts.message(&Alert { topic }).publish().await.is_err() {
110///         return HandlerOutcome::retry();
111///     }
112///     HandlerOutcome::ack()
113/// }
114///
115/// pub fn app() -> impl App {
116///     RustStream::new(AppInfo::new("telemetry", "0.1.0")).with_broker(
117///         MqttBroker::new("mqtt://localhost:1883", "telemetry-svc"),
118///         |b| {
119///             b.include(watch)
120///                 .out(DefaultSlot, Publish::default())
121///                 .out_retry(Publish::default())
122///                 .to("devices/retry/telemetry")
123///                 .build();
124///         },
125///     )
126/// }
127///
128/// pub async fn an_alert_names_the_topic_not_the_filter() -> Result<(), Box<dyn Error>> {
129///     let tb = TestApp::start(app()).await?;
130///
131///     tb.broker::<MqttBroker>()
132///         .message(&Telemetry { temperature: 31.5 })
133///         .to("devices/dev42/telemetry")
134///         .publish()
135///         .await?;
136///
137///     tb.broker::<MqttBroker>()
138///         .published::<Alert>("alerts")
139///         .assert_called_once()
140///         .with(&Alert { topic: "devices/dev42/telemetry".to_owned() });
141///     tb.shutdown().await?;
142///     Ok(())
143/// }
144/// # }
145/// # #[cfg(feature = "testing")]
146/// # fn main() -> Result<(), Box<dyn std::error::Error>> {
147/// #     tokio::runtime::Builder::new_multi_thread()
148/// #         .enable_all()
149/// #         .build()?
150/// #         .block_on(demo::an_alert_names_the_topic_not_the_filter())
151/// # }
152/// # #[cfg(not(feature = "testing"))]
153/// # fn main() {}
154/// ```
155#[derive(Debug, Default, Clone, Copy, PartialEq, Eq)]
156pub struct DeliveryTopic;
157
158impl Field<MqttContext> for DeliveryTopic {
159    type Value<'a> = &'a str;
160
161    fn get(self, src: &MqttContext) -> &str {
162        src.topic()
163    }
164}
165
166impl ContextField for DeliveryTopic {
167    type Context = MqttContext;
168    type Value = String;
169
170    fn read(self, src: &MqttContext) -> String {
171        src.topic.clone()
172    }
173}
174
175impl BuildContext<MqttMessage> for MqttContext {
176    fn build(msg: &MqttMessage) -> Self {
177        Self::new(msg.topic())
178    }
179}
ruststream_rumqttc/context.rs

ruststream_rumqttc/
context.rs
