ruststream_rumqttc/
publisher.rs

1//! [`MqttPublisher`] and its [`MqttPublish`] policy.
2
3use bytes::Bytes;
4use rumqttc::v5::mqttbytes::valid_topic;
5use ruststream::{OutgoingMessage, PairError, PublishPolicy, Publisher};
6
7use crate::broker::{ConnectedMqttBroker, CoreCell};
8use crate::error::MqttError;
9use crate::filter::Qos;
10use crate::message::to_publish_properties;
11
12/// Publishes messages to MQTT topics through the shared connection.
13///
14/// The publish is queued into the client session: for `QoS` 1/2 the session's state machine
15/// retransmits until the broker acknowledges (surviving reconnects), so `Ok` means "owned by
16/// the session", not "broker confirmed". Buildable before `connect` and usable until
17/// `shutdown`; afterwards every publish reports [`MqttError::NotConnected`].
18#[derive(Clone)]
19pub struct MqttPublisher {
20    cell: CoreCell,
21    qos: Qos,
22    retain: bool,
23}
24
25impl std::fmt::Debug for MqttPublisher {
26    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
27        f.debug_struct("MqttPublisher")
28            .field("qos", &self.qos)
29            .field("retain", &self.retain)
30            .finish_non_exhaustive()
31    }
32}
33
34impl MqttPublisher {
35    pub(crate) fn new(cell: CoreCell, qos: Qos, retain: bool) -> Self {
36        Self { cell, qos, retain }
37    }
38}
39
40impl Publisher for MqttPublisher {
41    type Error = MqttError;
42
43    async fn publish(&self, msg: OutgoingMessage<'_>) -> Result<(), Self::Error> {
44        let core = self.cell.get().ok_or(MqttError::NotConnected)?;
45        core.shared.ensure_open()?;
46        // The client's send-path error cannot say why a request failed, so the topic is
47        // validated here; a remaining failure unambiguously means the connection is gone.
48        if !valid_topic(msg.name()) {
49            return Err(MqttError::Publish {
50                topic: msg.name().to_owned(),
51                reason: "not a valid MQTT topic (wildcards are subscribe-only)".to_owned(),
52            });
53        }
54        let payload = Bytes::copy_from_slice(msg.payload());
55        let outcome = match to_publish_properties(&msg) {
56            Some(properties) => {
57                core.client
58                    .publish_bytes_with_properties(
59                        msg.name(),
60                        self.qos.to_client(),
61                        self.retain,
62                        payload,
63                        properties,
64                    )
65                    .await
66            }
67            None => {
68                core.client
69                    .publish_bytes(msg.name(), self.qos.to_client(), self.retain, payload)
70                    .await
71            }
72        };
73        outcome.map_err(|_| MqttError::Publish {
74            topic: msg.name().to_owned(),
75            reason: "the mqtt connection task has shut down".to_owned(),
76        })
77    }
78}
79
80/// The publish policy for [`MqttPublisher`]: quality of service and the retain flag as pure
81/// declaration, paired with the connected broker by the runtime after `connect`.
82///
83/// # Examples
84///
85/// ```
86/// use ruststream_rumqttc::{MqttPublish, Qos};
87///
88/// let policy = MqttPublish::default().qos(Qos::ExactlyOnce).retain(true);
89/// # let _ = policy;
90/// ```
91#[derive(Debug, Clone, Copy, Default)]
92#[must_use]
93pub struct MqttPublish {
94    qos: Qos,
95    retain: bool,
96}
97
98impl MqttPublish {
99    /// Sets the delivery quality of service. Defaults to [`Qos::AtLeastOnce`].
100    pub fn qos(mut self, qos: Qos) -> Self {
101        self.qos = qos;
102        self
103    }
104
105    /// Publishes messages as retained: the broker keeps the last one per topic and hands it
106    /// to new (non-shared) subscribers.
107    pub fn retain(mut self, retain: bool) -> Self {
108        self.retain = retain;
109        self
110    }
111}
112
113impl MqttPublish {
114    pub(crate) fn into_publisher(self, cell: CoreCell) -> MqttPublisher {
115        MqttPublisher::new(cell, self.qos, self.retain)
116    }
117}
118
119impl PublishPolicy<ConnectedMqttBroker> for MqttPublish {
120    type Live = MqttPublisher;
121
122    async fn pair(self, connected: &ConnectedMqttBroker) -> Result<Self::Live, PairError> {
123        Ok(connected.publisher_with(self))
124    }
125}
ruststream_rumqttc/publisher.rs

ruststream_rumqttc/
publisher.rs
