#
google_cloud_pubsub/subscriber/
builder.rs

    1// Copyright 2025 Google LLC
    2//
    3// Licensed under the Apache License, Version 2.0 (the "License");
    4// you may not use this file except in compliance with the License.
    5// You may obtain a copy of the License at
    6//
    7//     https://www.apache.org/licenses/LICENSE-2.0
    8//
    9// Unless required by applicable law or agreed to in writing, software
    10// distributed under the License is distributed on an "AS IS" BASIS,
    11// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    12// See the License for the specific language governing permissions and
    13// limitations under the License.
    14
    15use super::session::Session;
    16use super::transport::Transport;
    17use crate::Result;
    18use std::sync::Arc;
    19
    20const MIB: i64 = 1024 * 1024;
    21
    22/// Builder for the `client::Subscriber::streaming_pull` method.
    23pub struct StreamingPull {
    24    pub(super) inner: Arc<Transport>,
    25    pub(super) subscription: String,
    26    pub(super) ack_deadline_seconds: i32,
    27    pub(super) max_outstanding_messages: i64,
    28    pub(super) max_outstanding_bytes: i64,
    29}
    30
    31impl StreamingPull {
    32    pub(super) fn new(inner: Arc<Transport>, subscription: String) -> Self {
    33        Self {
    34            inner,
    35            subscription,
    36            ack_deadline_seconds: 10,
    37            max_outstanding_messages: 1000,
    38            max_outstanding_bytes: 100 * MIB,
    39        }
    40    }
    41
    42    /// Creates a new session to receive messages from the subscription.
    43    ///
    44    /// # Example
    45    /// ```
    46    /// # use google_cloud_pubsub::client::Subscriber;
    47    /// # async fn sample(client: Subscriber) -> anyhow::Result<()> {
    48    /// let mut session = client
    49    ///     .streaming_pull("projects/my-project/subscriptions/my-subscription")
    50    ///     .start()
    51    ///     .await?;
    52    /// while let Some((m, h)) = session.next().await.transpose()? {
    53    ///     println!("Received message m={m:?}");
    54    ///     h.ack();
    55    /// }
    56    /// # Ok(()) }
    57    /// ```
    58    pub async fn start(self) -> Result<Session> {
    59        Session::new(self).await
    60    }
    61
    62    /// Sets the ack deadline to use for the stream.
    63    ///
    64    /// This value represents how long the application has to ack or nack an
    65    /// incoming message. Note that this value is independent of the deadline
    66    /// configured on the server-side subscription.
    67    ///
    68    /// If the server does not hear back from the client within this deadline
    69    /// (e.g. if an application crashes), it will resend any unacknowledged
    70    /// messages to another subscriber.
    71    ///
    72    /// The minimum deadline you can specify is 10 seconds. The maximum deadline
    73    /// you can specify is 600 seconds (10 minutes).
    74    ///
    75    /// The default value is 10 seconds.
    76    ///
    77    /// # Example
    78    /// ```
    79    /// # use google_cloud_pubsub::client::Subscriber;
    80    /// # async fn sample() -> anyhow::Result<()> {
    81    /// # let client = Subscriber::builder().build().await?;
    82    /// let session = client.streaming_pull("projects/my-project/subscriptions/my-subscription")
    83    ///     .set_ack_deadline_seconds(20)
    84    ///     .start();
    85    /// # Ok(()) }
    86    /// ```
    87    pub fn set_ack_deadline_seconds<T: Into<i32>>(mut self, v: T) -> Self {
    88        self.ack_deadline_seconds = v.into();
    89        self
    90    }
    91
    92    /// Flow control settings for the maximum number of outstanding messages.
    93    ///
    94    /// The server will stop sending messages to a client when this many
    95    /// messages are outstanding (i.e. that have not been acked or nacked).
    96    ///
    97    /// The server resumes sending messages when the outstanding message count
    98    /// drops below this value.
    99    ///
    100    /// Use a value <= 0 to set no limit on the number of outstanding messages.
    101    ///
    102    /// The default value is 1000 messages.
    103    ///
    104    /// # Example
    105    /// ```
    106    /// # use google_cloud_pubsub::client::Subscriber;
    107    /// # async fn sample() -> anyhow::Result<()> {
    108    /// # let client = Subscriber::builder().build().await?;
    109    /// let session = client.streaming_pull("projects/my-project/subscriptions/my-subscription")
    110    ///     .set_max_outstanding_messages(2000)
    111    ///     .start();
    112    /// # Ok(()) }
    113    /// ```
    114    pub fn set_max_outstanding_messages<T: Into<i64>>(mut self, v: T) -> Self {
    115        self.max_outstanding_messages = v.into();
    116        self
    117    }
    118
    119    /// Flow control settings for the maximum number of outstanding bytes.
    120    ///
    121    /// The server will stop sending messages to a client when this many bytes
    122    /// of messages are outstanding (i.e. that have not been acked or nacked).
    123    ///
    124    /// The server resumes sending messages when the outstanding byte count
    125    /// drops below this value.
    126    ///
    127    /// Use a value <= 0 to set no limit on the number of outstanding bytes.
    128    ///
    129    /// The default value is 100 MiB.
    130    ///
    131    /// # Example
    132    /// ```no_rust
    133    /// # use google_cloud_pubsub::client::Subscriber;
    134    /// # async fn sample() -> anyhow::Result<()> {
    135    /// # let client = Subscriber::builder().build().await?;
    136    /// const MIB: i64 = 1024 * 1024;
    137    /// let session = client.streaming_pull("projects/my-project/subscriptions/my-subscription")
    138    ///     .set_max_outstanding_bytes(200 * MIB)
    139    ///     .start();
    140    /// # Ok(()) }
    141    /// ```
    142    pub fn set_max_outstanding_bytes<T: Into<i64>>(mut self, v: T) -> Self {
    143        self.max_outstanding_bytes = v.into();
    144        self
    145    }
    146}
    147
    148#[cfg(test)]
    149mod tests {
    150    use super::*;
    151    use auth::credentials::anonymous::Builder as Anonymous;
    152    use gaxi::options::ClientConfig;
    153
    154    const KIB: i64 = 1024;
    155
    156    async fn test_inner() -> anyhow::Result<Arc<Transport>> {
    157        let mut config = ClientConfig::default();
    158        config.cred = Some(Anonymous::new().build());
    159        let transport = Transport::new(config).await?;
    160        Ok(Arc::new(transport))
    161    }
    162
    163    #[tokio::test]
    164    async fn reasonable_defaults() -> anyhow::Result<()> {
    165        let builder = StreamingPull::new(
    166            test_inner().await?,
    167            "projects/my-project/subscriptions/my-subscription".to_string(),
    168        );
    169        assert_eq!(
    170            builder.subscription,
    171            "projects/my-project/subscriptions/my-subscription"
    172        );
    173        assert_eq!(builder.ack_deadline_seconds, 10);
    174        assert!(
    175            100_000 > builder.max_outstanding_messages && builder.max_outstanding_messages > 100,
    176            "max_outstanding_messages={}",
    177            builder.max_outstanding_messages
    178        );
    179        assert!(
    180            builder.max_outstanding_bytes > 100 * KIB,
    181            "max_outstanding_bytes={}",
    182            builder.max_outstanding_bytes
    183        );
    184
    185        Ok(())
    186    }
    187
    188    #[tokio::test]
    189    async fn options() -> anyhow::Result<()> {
    190        let builder = StreamingPull::new(
    191            test_inner().await?,
    192            "projects/my-project/subscriptions/my-subscription".to_string(),
    193        )
    194        .set_ack_deadline_seconds(20)
    195        .set_max_outstanding_messages(12345)
    196        .set_max_outstanding_bytes(6789 * KIB);
    197        assert_eq!(
    198            builder.subscription,
    199            "projects/my-project/subscriptions/my-subscription"
    200        );
    201        assert_eq!(builder.ack_deadline_seconds, 20);
    202        assert_eq!(builder.max_outstanding_messages, 12345);
    203        assert_eq!(builder.max_outstanding_bytes, 6789 * KIB);
    204
    205        Ok(())
    206    }
    207}