google_pubsub 0.7.8

pubsub
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69

use crate::api::{self, Topic as APITopic};
use crate::client::Client;
use crate::error::Error;
use crate::Subscription;
use std::collections::HashMap;

#[derive(Clone)]
pub struct Topic {
    pub(crate) client: Client,
    pub(crate) name: String,
}

impl Topic {
    pub(crate) fn new(client: Client, name: impl Into<String>) -> Self {
        Self {
            client,
            name: name.into(),
        }
    }
    pub async fn create_topic(&mut self) -> Result<(), Error> {
        let mut topic = APITopic::default();
        topic.name = self.name.clone();
        let req = self.client.insert_authz_token(topic).await?;
        self.client.publisher.create_topic(req).await?;
        Ok(())
    }
    pub async fn create_subscription(
        &mut self,
        subscription_name: &str,
        filter: &str,
    ) -> Result<Subscription, Error> {
        let resource_name = format!(
            "projects/{}/subscriptions/{}",
            &self.client.project_id, subscription_name
        );
        let subs = api::Subscription {
            name: resource_name.clone(),
            ack_deadline_seconds: 10,
            dead_letter_policy: None,
            detached: false,
            enable_message_ordering: true,
            expiration_policy: None,
            filter: filter.to_string(),
            labels: HashMap::new(),
            message_retention_duration: None,
            push_config: None,
            retain_acked_messages: false,
            retry_policy: None,
            topic: self.name.clone(),
        };
        let req = self.client.insert_authz_token(subs).await?;
        self.client.subscriber.create_subscription(req).await?;
        Ok(Subscription::new(self.client.clone(), &resource_name))
    }
    pub async fn delete_subscription(&mut self, subscription_name: &str) -> Result<(), Error> {
        let delete_sub_req = api::DeleteSubscriptionRequest {
            subscription: format!(
                "projects/{}/subscriptions/{}",
                &self.client.project_id, subscription_name
            ),
        };
        let authorized_req = self.client.insert_authz_token(delete_sub_req).await?;
        self.client
            .subscriber
            .delete_subscription(authorized_req)
            .await?;
        Ok(())
    }
}