gcloud-pubsub 1.7.0

Google Cloud Platform pubsub client library.
Documentation 
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
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134

use std::sync::Arc;

use google_cloud_gax::conn::Channel;
use google_cloud_gax::create_request;
use google_cloud_gax::grpc::Response;
use google_cloud_gax::grpc::Status;
use google_cloud_gax::retry::{invoke, MapErr, RetrySetting};
use google_cloud_googleapis::pubsub::v1::schema_service_client::SchemaServiceClient;
use google_cloud_googleapis::pubsub::v1::{
    CreateSchemaRequest, DeleteSchemaRequest, GetSchemaRequest, ListSchemasRequest, Schema, ValidateMessageRequest,
    ValidateMessageResponse, ValidateSchemaRequest, ValidateSchemaResponse,
};

use crate::apiv1::conn_pool::ConnectionManager;

#[derive(Clone, Debug)]
pub(crate) struct SchemaClient {
    cm: Arc<ConnectionManager>,
}

#[allow(dead_code)]
impl SchemaClient {
    /// create new publisher client
    pub fn new(cm: ConnectionManager) -> SchemaClient {
        SchemaClient { cm: Arc::new(cm) }
    }

    fn client(&self) -> SchemaServiceClient<Channel> {
        SchemaServiceClient::new(self.cm.conn())
    }

    /// create_schema creates a schema.
    pub async fn create_schema(
        &self,
        req: CreateSchemaRequest,
        retry: Option<RetrySetting>,
    ) -> Result<Response<Schema>, Status> {
        let parent = &req.parent;
        let action = || async {
            let mut client = self.client();
            let request = create_request(format!("parent={parent}"), req.clone());
            client.create_schema(request).await.map_transient_err()
        };
        invoke(retry, action).await
    }

    /// get_schema gets a schema.
    pub async fn get_schema(
        &self,
        req: GetSchemaRequest,
        retry: Option<RetrySetting>,
    ) -> Result<Response<Schema>, Status> {
        let name = &req.name;
        let action = || async {
            let mut client = self.client();
            let request = create_request(format!("name={name}"), req.clone());
            client.get_schema(request).await.map_transient_err()
        };
        invoke(retry, action).await
    }

    /// list_schemas lists matching topics.
    pub async fn list_schemas(
        &self,
        mut req: ListSchemasRequest,
        retry: Option<RetrySetting>,
    ) -> Result<Vec<Schema>, Status> {
        let project = &req.parent;
        let mut all = vec![];
        //eager loading
        loop {
            let action = || async {
                let mut client = self.client();
                let request = create_request(format!("project={project}"), req.clone());
                client
                    .list_schemas(request)
                    .await
                    .map(|d| d.into_inner())
                    .map_transient_err()
            };
            let response = invoke(retry.clone(), action).await?;
            all.extend(response.schemas.into_iter());
            if response.next_page_token.is_empty() {
                return Ok(all);
            }
            req.page_token = response.next_page_token;
        }
    }

    /// delete_schema deletes a schema.
    pub async fn delete_schema(
        &self,
        req: DeleteSchemaRequest,
        retry: Option<RetrySetting>,
    ) -> Result<Response<()>, Status> {
        let name = &req.name;
        let action = || async {
            let mut client = self.client();
            let request = create_request(format!("name={name}"), req.clone());
            client.delete_schema(request).await.map_transient_err()
        };
        invoke(retry, action).await
    }

    /// validate_schema deletes a schema.
    pub async fn validate_schema(
        &self,
        req: ValidateSchemaRequest,
        retry: Option<RetrySetting>,
    ) -> Result<Response<ValidateSchemaResponse>, Status> {
        let parent = &req.parent;
        let action = || async {
            let mut client = self.client();
            let request = create_request(format!("parent={parent}"), req.clone());
            client.validate_schema(request).await.map_transient_err()
        };
        invoke(retry, action).await
    }

    /// validate_message validates a message against a schema.
    pub async fn validate_message(
        &self,
        req: ValidateMessageRequest,
        retry: Option<RetrySetting>,
    ) -> Result<Response<ValidateMessageResponse>, Status> {
        let parent = &req.parent;
        let action = || async {
            let mut client = self.client();
            let request = create_request(format!("parent={parent}"), req.clone());
            client.validate_message(request).await.map_transient_err()
        };
        invoke(retry, action).await
    }
}