#
google_cloud_pubsub/subscriber/
client.rs

    1// Copyright 2026 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::builder::StreamingPull;
    16use super::client_builder::ClientBuilder;
    17use super::transport::Transport;
    18use gax::client_builder::Result as BuilderResult;
    19use std::sync::Arc;
    20
    21/// A Subscriber client for the [Cloud Pub/Sub] API.
    22///
    23/// Use this client to receive messages from a [pull subscription] on a topic.
    24///
    25/// # Example
    26/// ```
    27/// # use google_cloud_pubsub::client::Subscriber;
    28/// # async fn sample() -> anyhow::Result<()> {
    29/// let client = Subscriber::builder().build().await?;
    30/// let mut session = client
    31///     .streaming_pull("projects/my-project/subscriptions/my-subscription")
    32///     .start()
    33///     .await?;
    34/// while let Some((m, h)) = session.next().await.transpose()? {
    35///     println!("Received message m={m:?}");
    36///     h.ack();
    37/// }
    38/// # Ok(()) }
    39/// ```
    40///
    41/// # Configuration
    42///
    43/// To configure a `Subscriber` use the `with_*` methods in the type returned by
    44/// [builder()][Subscriber::builder]. The default configuration should work for
    45/// most applications. Common configuration changes include:
    46///
    47/// * [with_endpoint()]: by default this client uses the global default endpoint
    48///   (`https://pubsub.googleapis.com`). Applications using regional endpoints
    49///   or running in restricted networks (e.g. a network configured with
    50///   [Private Google Access with VPC Service Controls]) may want to override
    51///   this default.
    52/// * [with_credentials()]: by default this client uses
    53///   [Application Default Credentials]. Applications using custom
    54///   authentication may need to override this default.
    55///
    56/// # Pooling and Cloning
    57///
    58/// `Subscriber` holds a connection pool internally, it is advised to
    59/// create one and then reuse it.  You do not need to wrap `Subscriber` in
    60/// an [Rc](std::rc::Rc) or [Arc] to reuse it, because it already uses an `Arc`
    61/// internally.
    62///
    63/// [application default credentials]: https://cloud.google.com/docs/authentication#adc
    64/// [cloud pub/sub]: https://docs.cloud.google.com/pubsub/docs/overview
    65/// [private google access with vpc service controls]: https://cloud.google.com/vpc-service-controls/docs/private-connectivity
    66/// [pull subscription]: https://docs.cloud.google.com/pubsub/docs/pull
    67/// [with_endpoint()]: ClientBuilder::with_endpoint
    68/// [with_credentials()]: ClientBuilder::with_credentials
    69#[derive(Clone, Debug)]
    70pub struct Subscriber {
    71    inner: Arc<Transport>,
    72}
    73
    74impl Subscriber {
    75    /// Returns a builder for [Subscriber].
    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    /// # Ok(()) }
    83    /// ```
    84    pub fn builder() -> ClientBuilder {
    85        ClientBuilder::new()
    86    }
    87
    88    /// Receive messages from a [subscription].
    89    ///
    90    /// The `subscription` is the full name, in the format of
    91    /// `projects/*/subscriptions/*`.
    92    ///
    93    /// # Example
    94    /// ```
    95    /// # use google_cloud_pubsub::client::Subscriber;
    96    /// # async fn sample(client: Subscriber) -> anyhow::Result<()> {
    97    /// let mut session = client
    98    ///     .streaming_pull("projects/my-project/subscriptions/my-subscription")
    99    ///     .start()
    100    ///     .await?;
    101    /// while let Some((m, h)) = session.next().await.transpose()? {
    102    ///     println!("Received message m={m:?}");
    103    ///     h.ack();
    104    /// }
    105    /// # Ok(()) }
    106    /// ```
    107    ///
    108    /// [subscription]: https://docs.cloud.google.com/pubsub/docs/subscription-overview
    109    pub fn streaming_pull<T>(&self, subscription: T) -> StreamingPull
    110    where
    111        T: Into<String>,
    112    {
    113        StreamingPull::new(self.inner.clone(), subscription.into())
    114    }
    115
    116    pub(super) async fn new(builder: ClientBuilder) -> BuilderResult<Self> {
    117        let transport = Transport::new(builder.config).await?;
    118        Ok(Self {
    119            inner: Arc::new(transport),
    120        })
    121    }
    122}
    123
    124#[cfg(test)]
    125mod tests {
    126    use super::*;
    127    use auth::credentials::anonymous::Builder as Anonymous;
    128    use pubsub_grpc_mock::{MockSubscriber, start};
    129
    130    #[tokio::test]
    131    async fn basic() -> anyhow::Result<()> {
    132        let _ = Subscriber::builder().build().await?;
    133        Ok(())
    134    }
    135
    136    #[tokio::test]
    137    async fn streaming_pull() -> anyhow::Result<()> {
    138        let mut mock = MockSubscriber::new();
    139        mock.expect_streaming_pull()
    140            .return_once(|_| Err(tonic::Status::internal("fail")));
    141        let (endpoint, _server) = start("0.0.0.0:0", mock).await?;
    142        let client = Subscriber::builder()
    143            .with_endpoint(endpoint)
    144            .with_credentials(Anonymous::new().build())
    145            .build()
    146            .await?;
    147        let err = client
    148            .streaming_pull("projects/p/subscriptions/s")
    149            .start()
    150            .await
    151            .expect_err("Session should not be created.");
    152        assert!(err.status().is_some(), "{err:?}");
    153        let status = err.status().unwrap();
    154        assert_eq!(status.code, gax::error::rpc::Code::Internal);
    155        assert_eq!(status.message, "fail");
    156
    157        Ok(())
    158    }
    159}