ruststream_gcp_pubsub/
subscription.rs

1//! [`PubSubSubscription`]: the subscription descriptor.
2//!
3//! Pub/Sub separates the topic from the subscription, and the descriptor keeps both explicit:
4//! by default it names an existing subscription; `create_with_topic` opts into creating the
5//! subscription (and its topic) on subscribe, which is what local development against the
6//! emulator wants.
7
8use std::time::Duration;
9
10use ruststream::SubscriptionSource;
11
12use crate::broker::ConnectedPubSubBroker;
13use crate::error::PubSubError;
14use crate::subscriber::PubSubSubscriber;
15
16/// A subscription descriptor for one Pub/Sub subscription.
17///
18/// Implements [`SubscriptionSource`], so it can sit inline in the `#[subscriber(..)]`
19/// decorator:
20///
21/// ```
22/// use std::time::Duration;
23/// use ruststream_gcp_pubsub::PubSubSubscription;
24///
25/// let source = PubSubSubscription::new("orders-workers")
26///     .max_outstanding(1_000)
27///     .ack_extension(Duration::from_secs(60));
28/// # let _ = source;
29/// ```
30#[derive(Debug, Clone, PartialEq, Eq)]
31#[must_use]
32pub struct PubSubSubscription {
33    name: String,
34    create_with_topic: Option<String>,
35    max_outstanding: Option<i64>,
36    ack_extension: Option<Duration>,
37}
38
39impl PubSubSubscription {
40    /// Names an existing subscription (short name or full
41    /// `projects/{p}/subscriptions/{s}` resource name).
42    pub fn new(name: impl Into<String>) -> Self {
43        Self {
44            name: name.into(),
45            create_with_topic: None,
46            max_outstanding: None,
47            ack_extension: None,
48        }
49    }
50
51    /// Creates the subscription bound to `topic` on subscribe when it does not exist yet (the
52    /// topic is created too). Meant for local development and tests against the emulator;
53    /// production subscriptions are usually managed as infrastructure.
54    pub fn create_with_topic(mut self, topic: impl Into<String>) -> Self {
55        self.create_with_topic = Some(topic.into());
56        self
57    }
58
59    /// Flow control: how many received messages may be outstanding (unacked) at once. Defaults
60    /// to the client's 1000.
61    pub fn max_outstanding(mut self, messages: i64) -> Self {
62        self.max_outstanding = Some(messages);
63        self
64    }
65
66    /// How far each background ack-deadline extension reaches while a handler runs. The client
67    /// clamps it to the protocol's 10s..=600s range; defaults to 60s.
68    pub fn ack_extension(mut self, extension: Duration) -> Self {
69        self.ack_extension = Some(extension);
70        self
71    }
72
73    /// The subscription name this descriptor resolves.
74    #[must_use]
75    pub fn subscription(&self) -> &str {
76        &self.name
77    }
78
79    pub(crate) fn create_topic_ref(&self) -> Option<&str> {
80        self.create_with_topic.as_deref()
81    }
82
83    pub(crate) fn max_outstanding_value(&self) -> Option<i64> {
84        self.max_outstanding
85    }
86
87    pub(crate) fn ack_extension_value(&self) -> Option<Duration> {
88        self.ack_extension
89    }
90
91    /// Rejects descriptors that cannot form a subscription, before any I/O.
92    pub(crate) fn validate(&self) -> Result<(), PubSubError> {
93        if self.name.is_empty() {
94            return Err(PubSubError::InvalidDescriptor(
95                "subscription name must be non-empty".into(),
96            ));
97        }
98        if self.create_with_topic.as_deref() == Some("") {
99            return Err(PubSubError::InvalidDescriptor(
100                "topic name must be non-empty".into(),
101            ));
102        }
103        Ok(())
104    }
105}
106
107impl SubscriptionSource<ConnectedPubSubBroker> for PubSubSubscription {
108    type Subscriber = PubSubSubscriber;
109
110    fn name(&self) -> &str {
111        self.subscription()
112    }
113
114    async fn subscribe(
115        self,
116        connected: &ConnectedPubSubBroker,
117    ) -> Result<PubSubSubscriber, PubSubError> {
118        connected.subscribe_descriptor(self).await
119    }
120}
121
122#[cfg(test)]
123mod tests {
124    use super::*;
125
126    #[test]
127    fn empty_subscription_name_is_rejected_before_io() {
128        assert!(matches!(
129            PubSubSubscription::new("").validate(),
130            Err(PubSubError::InvalidDescriptor(_))
131        ));
132    }
133
134    #[test]
135    fn empty_topic_name_is_rejected_before_io() {
136        assert!(matches!(
137            PubSubSubscription::new("s")
138                .create_with_topic("")
139                .validate(),
140            Err(PubSubError::InvalidDescriptor(_))
141        ));
142    }
143}
ruststream_gcp_pubsub/subscription.rs

ruststream_gcp_pubsub/
subscription.rs
