ruststream_sqs_sns/
queue.rs

1//! [`SqsQueue`]: the subscription descriptor.
2//!
3//! The polling parameters that decide cost and latency are explicit: `wait` (long polling),
4//! `batch` (messages per receive call), and `visibility` (the redelivery timeout the crate
5//! keeps extending while a handler holds a message).
6
7use std::time::Duration;
8
9use ruststream::SubscriptionSource;
10
11use crate::broker::ConnectedSqsBroker;
12use crate::error::SqsError;
13use crate::subscriber::SqsSubscriber;
14
15/// The protocol cap on long polling.
16const MAX_WAIT: Duration = Duration::from_secs(20);
17
18/// A subscription descriptor for one SQS queue.
19///
20/// Accepts a queue URL or a queue name (resolved through `GetQueueUrl` on subscribe).
21/// Implements [`SubscriptionSource`], so it can sit inline in the `#[subscriber(..)]`
22/// decorator:
23///
24/// ```
25/// use std::time::Duration;
26/// use ruststream_sqs_sns::SqsQueue;
27///
28/// let source = SqsQueue::new("orders")
29///     .wait(Duration::from_secs(20))
30///     .batch(10)
31///     .visibility(Duration::from_secs(30));
32/// # let _ = source;
33/// ```
34#[derive(Debug, Clone, PartialEq, Eq)]
35#[must_use]
36pub struct SqsQueue {
37    queue: String,
38    wait: Duration,
39    batch: i32,
40    visibility: Option<Duration>,
41    create_if_missing: bool,
42}
43
44impl SqsQueue {
45    /// Names the queue by URL (`https://sqs...`) or by name.
46    pub fn new(queue: impl Into<String>) -> Self {
47        Self {
48            queue: queue.into(),
49            wait: MAX_WAIT,
50            batch: 10,
51            visibility: None,
52            create_if_missing: false,
53        }
54    }
55
56    /// Long-polling wait per receive call. Defaults to the protocol maximum of 20 seconds;
57    /// values above it are rejected before any I/O.
58    pub fn wait(mut self, wait: Duration) -> Self {
59        self.wait = wait;
60        self
61    }
62
63    /// Messages per receive call (1..=10, the protocol cap). Defaults to 10.
64    pub fn batch(mut self, batch: i32) -> Self {
65        self.batch = batch;
66        self
67    }
68
69    /// The visibility timeout requested per receive; the crate extends it in the background
70    /// while a handler holds the message. Defaults to the queue's configured timeout.
71    pub fn visibility(mut self, visibility: Duration) -> Self {
72        self.visibility = Some(visibility);
73        self
74    }
75
76    /// Creates the queue on subscribe when it does not exist yet (a name ending in `.fifo`
77    /// creates a FIFO queue with content-based deduplication). Meant for local development and
78    /// tests; production queues are usually managed as infrastructure.
79    pub fn create_if_missing(mut self) -> Self {
80        self.create_if_missing = true;
81        self
82    }
83
84    /// The queue URL or name this descriptor resolves.
85    #[must_use]
86    pub fn queue(&self) -> &str {
87        &self.queue
88    }
89
90    pub(crate) fn wait_value(&self) -> Duration {
91        self.wait
92    }
93
94    pub(crate) fn batch_value(&self) -> i32 {
95        self.batch
96    }
97
98    pub(crate) fn visibility_value(&self) -> Option<Duration> {
99        self.visibility
100    }
101
102    pub(crate) fn create_value(&self) -> bool {
103        self.create_if_missing
104    }
105
106    /// Rejects descriptors that cannot form a subscription, before any I/O.
107    pub(crate) fn validate(&self) -> Result<(), SqsError> {
108        if self.queue.is_empty() {
109            return Err(SqsError::InvalidQueue("queue must be non-empty".into()));
110        }
111        if self.wait > MAX_WAIT {
112            return Err(SqsError::InvalidQueue(
113                "wait exceeds the 20 second long-polling cap".into(),
114            ));
115        }
116        if !(1..=10).contains(&self.batch) {
117            return Err(SqsError::InvalidQueue(
118                "batch must be within 1..=10 (the receive cap)".into(),
119            ));
120        }
121        if let Some(visibility) = self.visibility
122            && (visibility.is_zero() || visibility > Duration::from_hours(12))
123        {
124            return Err(SqsError::InvalidQueue(
125                "visibility must be within 1s..=12h".into(),
126            ));
127        }
128        Ok(())
129    }
130}
131
132impl SubscriptionSource<ConnectedSqsBroker> for SqsQueue {
133    type Subscriber = SqsSubscriber;
134
135    fn name(&self) -> &str {
136        self.queue()
137    }
138
139    async fn subscribe(self, connected: &ConnectedSqsBroker) -> Result<SqsSubscriber, SqsError> {
140        connected.subscribe_queue(self).await
141    }
142}
143
144#[cfg(test)]
145mod tests {
146    use super::*;
147
148    #[test]
149    fn empty_queue_is_rejected_before_io() {
150        assert!(matches!(
151            SqsQueue::new("").validate(),
152            Err(SqsError::InvalidQueue(_))
153        ));
154    }
155
156    #[test]
157    fn overlong_wait_is_rejected_before_io() {
158        assert!(matches!(
159            SqsQueue::new("q").wait(Duration::from_secs(21)).validate(),
160            Err(SqsError::InvalidQueue(_))
161        ));
162    }
163
164    #[test]
165    fn out_of_range_batch_is_rejected_before_io() {
166        assert!(matches!(
167            SqsQueue::new("q").batch(11).validate(),
168            Err(SqsError::InvalidQueue(_))
169        ));
170        assert!(matches!(
171            SqsQueue::new("q").batch(0).validate(),
172            Err(SqsError::InvalidQueue(_))
173        ));
174    }
175}
ruststream_sqs_sns/queue.rs

ruststream_sqs_sns/
queue.rs
