ruststream_lapin/
subscriber.rs

1//! The subscriber: a stream of AMQP deliveries from one queue consumer.
2
3use futures::{Stream, StreamExt};
4use lapin::{Channel, Consumer};
5use ruststream::Subscriber;
6
7use crate::error::AmqpError;
8use crate::message::LapinMessage;
9
10/// A consumer on one queue, yielding [`LapinMessage`] deliveries.
11///
12/// Created by subscribing a [`RabbitQueue`](crate::RabbitQueue) descriptor (or a bare queue name)
13/// through [`LapinBroker`](crate::LapinBroker). The subscriber owns a dedicated channel;
14/// dropping it closes that channel and the broker redelivers whatever was unacknowledged.
15///
16/// Back-pressure: the broker stops pushing once
17/// [`prefetch`](crate::LapinBroker::prefetch) unacknowledged deliveries are in flight, so
18/// consuming slower slows the producer side down instead of buffering without bound.
19pub struct LapinSubscriber {
20    // Kept alive for the lifetime of the subscription: dropping the channel cancels the
21    // consumer server-side.
22    _channel: Channel,
23    consumer: Consumer,
24    queue: String,
25}
26
27impl LapinSubscriber {
28    pub(crate) fn new(channel: Channel, consumer: Consumer, queue: String) -> Self {
29        Self {
30            _channel: channel,
31            consumer,
32            queue,
33        }
34    }
35
36    /// The queue this subscriber consumes from.
37    #[must_use]
38    pub fn queue(&self) -> &str {
39        &self.queue
40    }
41}
42
43impl std::fmt::Debug for LapinSubscriber {
44    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
45        f.debug_struct("LapinSubscriber")
46            .field("queue", &self.queue)
47            .finish_non_exhaustive()
48    }
49}
50
51impl Subscriber for LapinSubscriber {
52    type Message = LapinMessage;
53    type Error = AmqpError;
54
55    /// Streams deliveries as they arrive; the stream ends when the consumer is cancelled or the
56    /// connection closes.
57    ///
58    /// # Cancel safety
59    ///
60    /// Polling is cancel safe (no delivery is lost by dropping the stream between polls), and
61    /// the stream can be re-created by calling `stream` again: deliveries buffer in the
62    /// consumer, not in the returned stream.
63    fn stream(&mut self) -> impl Stream<Item = Result<Self::Message, Self::Error>> + Send + '_ {
64        futures::stream::unfold(&mut self.consumer, |consumer| async move {
65            let item = consumer.next().await?;
66            let mapped = match item {
67                Ok(delivery) => Ok(LapinMessage::from_delivery(delivery)),
68                Err(err) => Err(AmqpError::consume(err)),
69            };
70            Some((mapped, consumer))
71        })
72    }
73}
ruststream_lapin/subscriber.rs

ruststream_lapin/
subscriber.rs
