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

ruststream_lapin/
subscriber.rs
