ruststream_rumqttc/
subscriber.rs

1//! [`MqttSubscriber`]: a stream of deliveries fed by the connection task.
2
3use std::sync::Arc;
4
5use futures::Stream;
6use rumqttc::v5::AsyncClient;
7use ruststream::Subscriber;
8use tokio::sync::mpsc;
9
10use crate::conn::Shared;
11use crate::error::MqttError;
12use crate::message::MqttMessage;
13
14/// A subscription to one MQTT topic filter; yields [`MqttMessage`]s.
15///
16/// Delivery back-pressure is the protocol's receive-maximum: the broker bounds unacked
17/// `QoS` 1/2 deliveries, so unsettled messages cap what sits in this subscriber's queue
18/// (`QoS` 0 has no such bound by design). Dropping the subscriber unsubscribes the filter.
19pub struct MqttSubscriber {
20    filter: String,
21    id: u64,
22    shared: Arc<Shared>,
23    client: AsyncClient,
24    rx: mpsc::UnboundedReceiver<Result<MqttMessage, MqttError>>,
25}
26
27impl std::fmt::Debug for MqttSubscriber {
28    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
29        f.debug_struct("MqttSubscriber")
30            .field("filter", &self.filter)
31            .finish_non_exhaustive()
32    }
33}
34
35impl MqttSubscriber {
36    pub(crate) fn new(
37        filter: String,
38        id: u64,
39        shared: Arc<Shared>,
40        client: AsyncClient,
41        rx: mpsc::UnboundedReceiver<Result<MqttMessage, MqttError>>,
42    ) -> Self {
43        Self {
44            filter,
45            id,
46            shared,
47            client,
48            rx,
49        }
50    }
51
52    /// The plain topic filter this subscription matches.
53    #[must_use]
54    pub fn filter(&self) -> &str {
55        &self.filter
56    }
57}
58
59impl Drop for MqttSubscriber {
60    fn drop(&mut self) {
61        if let Some(wire_filter) = self.shared.remove(self.id) {
62            let _ = self.client.try_unsubscribe(wire_filter);
63        }
64    }
65}
66
67impl Subscriber for MqttSubscriber {
68    type Message = MqttMessage;
69    type Error = MqttError;
70
71    fn stream(&mut self) -> impl Stream<Item = Result<MqttMessage, MqttError>> + Send + '_ {
72        // Poll the channel in place rather than wrapping it in an owning stream, so `stream`
73        // can be called again after the returned stream is dropped (the runtime and the
74        // conformance helpers re-enter it per call).
75        futures::stream::poll_fn(move |cx| self.rx.poll_recv(cx))
76    }
77}
ruststream_rumqttc/subscriber.rs

ruststream_rumqttc/
subscriber.rs
