#
google_cloud_pubsub/subscriber/
transport.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::stub::{Stub, TonicStreaming};
    16use crate::Result;
    17use crate::generated::gapic_dataplane::stub::dynamic::Subscriber as GapicStub;
    18pub(super) use crate::generated::gapic_dataplane::transport::Subscriber as Transport;
    19use crate::google::pubsub::v1::{StreamingPullRequest, StreamingPullResponse};
    20use tokio::sync::mpsc::Receiver;
    21use tokio_stream::wrappers::ReceiverStream;
    22
    23mod info {
    24    const NAME: &str = env!("CARGO_PKG_NAME");
    25    const VERSION: &str = env!("CARGO_PKG_VERSION");
    26    lazy_static::lazy_static! {
    27        pub(super) static ref X_GOOG_API_CLIENT_HEADER: String = {
    28            let ac = gaxi::api_header::XGoogApiClient{
    29                name:          NAME,
    30                version:       VERSION,
    31                library_type:  gaxi::api_header::GCCL,
    32            };
    33            ac.grpc_header_value()
    34        };
    35    }
    36}
    37
    38impl TonicStreaming for tonic::Streaming<StreamingPullResponse> {
    39    async fn next_message(&mut self) -> tonic::Result<Option<StreamingPullResponse>> {
    40        self.message().await
    41    }
    42}
    43
    44#[async_trait::async_trait]
    45impl Stub for Transport {
    46    type Stream = tonic::codec::Streaming<StreamingPullResponse>;
    47    async fn streaming_pull(
    48        &self,
    49        request_rx: Receiver<StreamingPullRequest>,
    50        options: gax::options::RequestOptions,
    51    ) -> Result<tonic::Response<Self::Stream>> {
    52        let request = ReceiverStream::new(request_rx);
    53        let extensions = {
    54            let mut e = tonic::Extensions::new();
    55            e.insert(tonic::GrpcMethod::new(
    56                "google.pubsub.v1.Subscriber",
    57                "StreamingPull",
    58            ));
    59            e
    60        };
    61        let path =
    62            http::uri::PathAndQuery::from_static("/google.pubsub.v1.Subscriber/StreamingPull");
    63        self.inner
    64            .bidi_stream(
    65                extensions,
    66                path,
    67                request,
    68                options,
    69                &info::X_GOOG_API_CLIENT_HEADER,
    70                "",
    71            )
    72            .await
    73    }
    74
    75    async fn modify_ack_deadline(
    76        &self,
    77        req: crate::model::ModifyAckDeadlineRequest,
    78        options: gax::options::RequestOptions,
    79    ) -> Result<gax::response::Response<()>> {
    80        GapicStub::modify_ack_deadline(self, req, options).await
    81    }
    82
    83    async fn acknowledge(
    84        &self,
    85        req: crate::model::AcknowledgeRequest,
    86        options: gax::options::RequestOptions,
    87    ) -> Result<gax::response::Response<()>> {
    88        GapicStub::acknowledge(self, req, options).await
    89    }
    90}
    91
    92#[cfg(test)]
    93pub(super) mod tests {
    94    use super::*;
    95    use crate::google::pubsub::v1::ReceivedMessage;
    96    use auth::credentials::anonymous::Builder as Anonymous;
    97    use pubsub_grpc_mock::google::pubsub::v1;
    98    use pubsub_grpc_mock::{MockSubscriber, start};
    99
    100    pub(in super::super) async fn test_transport(endpoint: String) -> anyhow::Result<Transport> {
    101        let mut config = gaxi::options::ClientConfig::default();
    102        config.cred = Some(Anonymous::new().build());
    103        config.endpoint = Some(endpoint);
    104        Ok(Transport::new(config).await?)
    105    }
    106
    107    // Both crates have their own copies of the protos. We can just serialize
    108    // then deserialize to convert between the two, as performance is not a
    109    // concern for these unit tests.
    110    fn convert(pb: &StreamingPullResponse) -> v1::StreamingPullResponse {
    111        use prost::Message;
    112        let v = pb.encode_to_vec();
    113        v1::StreamingPullResponse::decode(v.as_slice()).expect("encoding is always valid.")
    114    }
    115
    116    #[tokio::test]
    117    async fn streaming_pull() -> anyhow::Result<()> {
    118        let (response_tx, response_rx) = tokio::sync::mpsc::channel(1);
    119        let expected = StreamingPullResponse {
    120            received_messages: vec![ReceivedMessage {
    121                ack_id: "test-ack-id".to_string(),
    122                ..Default::default()
    123            }],
    124            ..Default::default()
    125        };
    126        response_tx.send(Ok(convert(&expected))).await?;
    127
    128        let (request_tx, request_rx) = tokio::sync::mpsc::channel(1);
    129        request_tx.send(StreamingPullRequest::default()).await?;
    130
    131        let mut mock = MockSubscriber::new();
    132        mock.expect_streaming_pull()
    133            .return_once(|_| Ok(tonic::Response::from(response_rx)));
    134        let (endpoint, _server) = start("0.0.0.0:0", mock).await?;
    135        let transport = test_transport(endpoint).await?;
    136        let mut stream = Stub::streaming_pull(
    137            &transport,
    138            request_rx,
    139            gax::options::RequestOptions::default(),
    140        )
    141        .await?
    142        .into_inner();
    143
    144        assert_eq!(stream.next_message().await?, Some(expected));
    145
    146        drop(response_tx);
    147        assert_eq!(stream.next_message().await?, None);
    148
    149        Ok(())
    150    }
    151
    152    #[tokio::test]
    153    async fn modify_ack_deadline() -> anyhow::Result<()> {
    154        let mut mock = MockSubscriber::new();
    155        mock.expect_modify_ack_deadline()
    156            .return_once(|_| Ok(tonic::Response::from(())));
    157        let (endpoint, _server) = start("0.0.0.0:0", mock).await?;
    158        let transport = test_transport(endpoint).await?;
    159        let _ = Stub::modify_ack_deadline(
    160            &transport,
    161            crate::model::ModifyAckDeadlineRequest::new(),
    162            gax::options::RequestOptions::default(),
    163        )
    164        .await?;
    165        Ok(())
    166    }
    167
    168    #[tokio::test]
    169    async fn acknowledge() -> anyhow::Result<()> {
    170        let mut mock = MockSubscriber::new();
    171        mock.expect_acknowledge()
    172            .return_once(|_| Ok(tonic::Response::from(())));
    173        let (endpoint, _server) = start("0.0.0.0:0", mock).await?;
    174        let transport = test_transport(endpoint).await?;
    175        let _ = Stub::acknowledge(
    176            &transport,
    177            crate::model::AcknowledgeRequest::new(),
    178            gax::options::RequestOptions::default(),
    179        )
    180        .await?;
    181        Ok(())
    182    }
    183}