Struct Subscription 

pub struct Subscription { /* private fields */ }

Subscription is a reference to a PubSub subscription.

Implementations

impl Subscription

pub fn id(&self) -> String

id returns the unique identifier of the subscription within its project.

pub fn fully_qualified_name(&self) -> &str

fully_qualified_name returns the globally unique printable name of the subscription.

pub fn fully_qualified_snapshot_name(&self, id: &str) -> String

fully_qualified_snapshot_name returns the globally unique printable name of the snapshot.

pub fn get_client(&self) -> SubscriberClient

pub async fn create( &self, fqtn: &str, cfg: SubscriptionConfig, retry: Option<RetrySetting>, ) -> Result<(), Status>

create creates the subscription.

pub async fn delete(&self, retry: Option<RetrySetting>) -> Result<(), Status>

delete deletes the subscription.

pub async fn exists(&self, retry: Option<RetrySetting>) -> Result<bool, Status>

exists reports whether the subscription exists on the server.

pub async fn config( &self, retry: Option<RetrySetting>, ) -> Result<(String, SubscriptionConfig), Status>

config fetches the current configuration for the subscription.

pub async fn update( &self, updating: SubscriptionConfigToUpdate, retry: Option<RetrySetting>, ) -> Result<(String, SubscriptionConfig), Status>

update changes an existing subscription according to the fields set in updating. It returns the new SubscriptionConfig.

pub async fn pull( &self, max_messages: i32, retry: Option<RetrySetting>, ) -> Result<Vec<ReceivedMessage>, Status>

pull get message synchronously. It blocks until at least one message is available.

pub async fn subscribe( &self, opt: Option<SubscribeConfig>, ) -> Result<MessageStream, Status>

subscribe creates a Stream of ReceivedMessage

use google_cloud_pubsub::subscription::{SubscribeConfig, Subscription};
use tokio::select;
use google_cloud_gax::grpc::Status;

async fn run(subscription: Subscription) -> Result<(), Status> {
    let mut iter = subscription.subscribe(None).await?;
    let ctx = iter.cancellable();
    let handler = tokio::spawn(async move {
        while let Some(message) = iter.read().await {
            let _ = message.ack().await;
        }
    });
    // Cancel and wait for nack all the pulled messages.
    ctx.cancel();
    let _ = handler.await;
    Ok(())
 }

use google_cloud_pubsub::subscription::{SubscribeConfig, Subscription};
use futures_util::StreamExt;
use tokio::select;
use google_cloud_gax::grpc::Status;

async fn run(subscription: Subscription) -> Result<(), Status> {
    let mut iter = subscription.subscribe(None).await?;
    let ctx = iter.cancellable();
    let handler = tokio::spawn(async move {
        while let Some(message) = iter.next().await {
            let _ = message.ack().await;
        }
    });
    // Cancel and wait for receive all the pulled messages.
    ctx.cancel();
    let _ = handler.await;
    Ok(())
 }

pub async fn receive<F>( &self, f: impl Fn(ReceivedMessage, CancellationToken) -> F + Send + 'static + Sync + Clone, cancel: CancellationToken, config: Option<ReceiveConfig>, ) -> Result<(), Status>

where F: Future<Output = ()> + Send + 'static,

receive calls f with the outstanding messages from the subscription. It blocks until cancellation token is cancelled, or the service returns a non-retryable error. The standard way to terminate a receive is to use CancellationToken.

pub async fn ack(&self, ack_ids: Vec<String>) -> Result<(), Status>

Ack acknowledges the messages associated with the ack_ids in the AcknowledgeRequest. The Pub/Sub system can remove the relevant messages from the subscription. This method is for batch acking.

use google_cloud_pubsub::client::Client;
use google_cloud_pubsub::subscription::Subscription;
use google_cloud_gax::grpc::Status;
use std::time::Duration;
use tokio_util::sync::CancellationToken;;

#[tokio::main]
async fn run(client: Client) -> Result<(), Status> {
    let subscription = client.subscription("test-subscription");
    let ctx = CancellationToken::new();
    let (sender, mut receiver)  = tokio::sync::mpsc::unbounded_channel();
    let subscription_for_receive = subscription.clone();
    let ctx_for_receive = ctx.clone();
    let ctx_for_ack_manager = ctx.clone();

    // receive
    let handle = tokio::spawn(async move {
        let _ = subscription_for_receive.receive(move |message, _ctx| {
            let sender = sender.clone();
            async move {
                let _ = sender.send(message.ack_id().to_string());
            }
        }, ctx_for_receive.clone(), None).await;
    });

    // batch ack manager
    let ack_manager = tokio::spawn( async move {
        let mut ack_ids = Vec::new();
        loop {
            tokio::select! {
                _ = ctx_for_ack_manager.cancelled() => {
                    return subscription.ack(ack_ids).await;
                },
                r = tokio::time::timeout(Duration::from_secs(10), receiver.recv()) => match r {
                    Ok(ack_id) => {
                        if let Some(ack_id) = ack_id {
                            ack_ids.push(ack_id);
                            if ack_ids.len() > 10 {
                                let _ = subscription.ack(ack_ids).await;
                                ack_ids = Vec::new();
                            }
                        }
                    },
                    Err(_e) => {
                        // timeout
                        let _ = subscription.ack(ack_ids).await;
                        ack_ids = Vec::new();
                    }
                }
            }
        }
    });

    ctx.cancel();
    Ok(())
 }

pub async fn seek( &self, to: SeekTo, retry: Option<RetrySetting>, ) -> Result<(), Status>

seek seeks the subscription a past timestamp or a saved snapshot.

pub async fn get_snapshot( &self, name: &str, retry: Option<RetrySetting>, ) -> Result<Snapshot, Status>

get_snapshot fetches an existing pubsub snapshot.

pub async fn create_snapshot( &self, name: &str, labels: HashMap<String, String>, retry: Option<RetrySetting>, ) -> Result<Snapshot, Status>

create_snapshot creates a new pubsub snapshot from the subscription’s state at the time of calling. The snapshot retains the messages for the topic the subscription is subscribed to, with the acknowledgment states consistent with the subscriptions. The created snapshot is guaranteed to retain:

• The message backlog on the subscription – or to be specific, messages that are unacknowledged at the time of the subscription’s creation.
• All messages published to the subscription’s topic after the snapshot’s creation. Snapshots have a finite lifetime – a maximum of 7 days from the time of creation, beyond which they are discarded and any messages being retained solely due to the snapshot dropped.

pub async fn delete_snapshot( &self, name: &str, retry: Option<RetrySetting>, ) -> Result<(), Status>

delete_snapshot deletes an existing pubsub snapshot.

Trait Implementations

impl Clone for Subscription

fn clone(&self) -> Subscription

Returns a duplicate of the value.

fn clone_from(&mut self, source: &Self)

Performs copy-assignment from source.

impl Debug for Subscription

fn fmt(&self, f: &mut Formatter<'_>) -> Result

Formats the value using the given formatter.