#
google_cloud_pubsub/subscriber/
stub.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 crate::Result;
    16use crate::google::pubsub::v1::{StreamingPullRequest, StreamingPullResponse};
    17use tokio::sync::mpsc::Receiver;
    18
    19pub(super) trait TonicStreaming: std::fmt::Debug + Send + 'static {
    20    fn next_message(
    21        &mut self,
    22    ) -> impl Future<Output = tonic::Result<Option<StreamingPullResponse>>> + Send;
    23}
    24
    25/// An internal trait for mocking the transport layer.
    26#[async_trait::async_trait]
    27pub(super) trait Stub: std::fmt::Debug + Send + Sync {
    28    type Stream: Sized;
    29    async fn streaming_pull(
    30        &self,
    31        request_rx: Receiver<StreamingPullRequest>,
    32        _options: gax::options::RequestOptions,
    33    ) -> Result<tonic::Response<Self::Stream>>;
    34
    35    async fn modify_ack_deadline(
    36        &self,
    37        _req: crate::model::ModifyAckDeadlineRequest,
    38        _options: gax::options::RequestOptions,
    39    ) -> Result<gax::response::Response<()>>;
    40
    41    async fn acknowledge(
    42        &self,
    43        _req: crate::model::AcknowledgeRequest,
    44        _options: gax::options::RequestOptions,
    45    ) -> Result<gax::response::Response<()>>;
    46}
    47
    48#[cfg(test)]
    49pub(super) mod tests {
    50    use super::*;
    51    use tokio::sync::mpsc::Receiver;
    52
    53    type MockStream = Receiver<tonic::Result<StreamingPullResponse>>;
    54
    55    // Allow us to mock a tonic stream in our unit tests, using an mpsc receiver
    56    impl TonicStreaming for MockStream {
    57        async fn next_message(&mut self) -> tonic::Result<Option<StreamingPullResponse>> {
    58            self.recv().await.transpose()
    59        }
    60    }
    61
    62    mockall::mock! {
    63        #[derive(Debug)]
    64        pub(in super::super) Stub {}
    65        #[async_trait::async_trait]
    66        impl Stub for Stub {
    67            type Stream = MockStream;
    68            async fn streaming_pull(
    69                &self,
    70                request_rx: Receiver<StreamingPullRequest>,
    71                _options: gax::options::RequestOptions,
    72            ) -> Result<tonic::Response<MockStream>>;
    73            async fn modify_ack_deadline(&self,
    74                _req: crate::model::ModifyAckDeadlineRequest,
    75                _options: gax::options::RequestOptions,
    76            ) -> Result<gax::response::Response<()>>;
    77            async fn acknowledge(
    78                &self,
    79                _req: crate::model::AcknowledgeRequest,
    80                _options: gax::options::RequestOptions,
    81            ) -> Result<gax::response::Response<()>>;
    82        }
    83    }
    84}