#
google_cloud_pubsub/publisher/
options.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
    15/// Configure publisher batching behavior.
    16#[derive(Clone, Debug)]
    17#[non_exhaustive]
    18pub struct BatchingOptions {
    19    pub message_count_threshold: u32,
    20    pub byte_threshold: u32,
    21    pub delay_threshold: std::time::Duration,
    22}
    23
    24impl BatchingOptions {
    25    /// Create a new instance.
    26    pub fn new() -> Self {
    27        Self::default()
    28    }
    29
    30    /// Set the [BatchingOptions][Self::message_count_threshold] field.
    31    pub fn set_message_count_threshold<V: Into<u32>>(mut self, v: V) -> Self {
    32        self.message_count_threshold = v.into();
    33        self
    34    }
    35
    36    /// Set the [BatchingOptions][Self::byte_threshold] field.
    37    pub(crate) fn set_byte_threshold<V: Into<u32>>(mut self, v: V) -> Self {
    38        self.byte_threshold = v.into();
    39        self
    40    }
    41
    42    /// Set the [BatchingOptions][Self::delay_threshold] field.
    43    pub fn set_delay_threshold<V: Into<std::time::Duration>>(mut self, v: V) -> Self {
    44        self.delay_threshold = v.into();
    45        self
    46    }
    47}
    48
    49impl std::default::Default for BatchingOptions {
    50    fn default() -> Self {
    51        Self {
    52            message_count_threshold: 100_u32,
    53            byte_threshold: 1000_u32,
    54            delay_threshold: std::time::Duration::from_millis(10),
    55        }
    56    }
    57}
    58
    59#[cfg(test)]
    60mod tests {
    61    use super::BatchingOptions;
    62
    63    #[tokio::test]
    64    async fn batching_options() -> anyhow::Result<()> {
    65        let options = BatchingOptions::new()
    66            .set_byte_threshold(1_234_u32)
    67            .set_message_count_threshold(123_u32)
    68            .set_delay_threshold(std::time::Duration::from_millis(12));
    69        assert_eq!(options.byte_threshold, 1_234_u32);
    70        assert_eq!(options.message_count_threshold, 123_u32);
    71        assert_eq!(
    72            options.delay_threshold,
    73            std::time::Duration::from_millis(12)
    74        );
    75        Ok(())
    76    }
    77}