ya-gcp 0.13.0

APIs for using Google Cloud Platform services
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
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152

//! An example of publishing data to pubsub and streaming back the same data

use futures::StreamExt;
use structopt::StructOpt;
use tracing_subscriber::{layer::SubscriberExt, util::SubscriberInitExt, EnvFilter, Registry};
use tracing_tree::HierarchicalLayer;

use ya_gcp::{pubsub, AuthFlow, ClientBuilder, ClientBuilderConfig, ServiceAccountAuth};

#[derive(Debug, StructOpt)]
struct Args {
    /// A path to the oauth service account key json file
    #[structopt(long)]
    service_account_key: std::path::PathBuf,

    /// The name of the pubsub project
    #[structopt(long)]
    project_name: String,

    /// The topic to create
    #[structopt(long)]
    topic: String,
}

#[tokio::main]
async fn main() -> Result<(), Box<dyn std::error::Error>> {
    let args = Args::from_args();

    Registry::default()
        .with(EnvFilter::from_default_env())
        .with(
            HierarchicalLayer::new(2)
                .with_targets(true)
                .with_bracketed_fields(true),
        )
        .init();

    println!("Building PubSub clients");

    let builder = ClientBuilder::new(ClientBuilderConfig::new().auth_flow(
        AuthFlow::ServiceAccount(ServiceAccountAuth::Path(args.service_account_key)),
    ))
    .await?;

    let pubsub_config = pubsub::PubSubConfig::default();

    // Prepare the publishing and subscribing clients

    let mut publisher = builder
        .build_pubsub_publisher(pubsub_config.clone())
        .await?;
    let mut subscriber = builder.build_pubsub_subscriber(pubsub_config).await?;

    // Make up topic and subscription names
    let topic_name = pubsub::ProjectTopicName::new(&args.project_name, &args.topic);
    let subscription_name = pubsub::ProjectSubscriptionName::new(&args.project_name, &args.topic);

    println!("Creating topic {}", &topic_name);

    publisher
        .raw_api_mut()
        .create_topic({
            let mut t = pubsub::api::Topic::default();
            t.name = topic_name.clone().into();
            t
        })
        .await?;

    println!("Creating subscription {}", &subscription_name);

    subscriber
        .raw_api_mut()
        .create_subscription({
            let mut s = pubsub::api::Subscription::default();
            s.name = subscription_name.clone().into();
            s.topic = topic_name.clone().into();
            s
        })
        .await?;

    println!("Publishing messages to topic");

    futures::stream::iter(0u32..15)
        .map(|i| {
            let mut m = pubsub::api::PubsubMessage::default();
            let payload = format!("message-{:02}", i);
            println!("Sending `{payload}`");
            m.data = payload.into();
            m
        })
        .map(Ok)
        .forward(publisher.publish_topic_sink(topic_name.clone(), pubsub::PublishConfig::default()))
        .await?;

    println!("Reading back published messages");

    let read_stream = subscriber.stream_subscription(
        subscription_name.clone(),
        pubsub::StreamSubscriptionConfig::default(),
    );
    futures::pin_mut!(read_stream);

    let mut messages = Vec::new();
    for _ in 0u32..15 {
        let (ack_token, message) = read_stream
            .next()
            .await
            .ok_or("unexpected end of stream")??;

        let payload = std::str::from_utf8(&message.data[..]).unwrap();
        println!("Received `{payload}`");
        messages.push(payload.to_owned());

        ack_token.ack().await?;
    }

    messages.sort();
    assert_eq!(
        messages,
        (0..15)
            .map(|i| format!("message-{:02}", i))
            .collect::<Vec<_>>()
    );

    println!("All messages matched!");

    println!("Deleting subscription {}", &subscription_name);

    subscriber
        .raw_api_mut()
        .delete_subscription({
            let mut d = pubsub::api::DeleteSubscriptionRequest::default();
            d.subscription = subscription_name.into();
            d
        })
        .await?;

    println!("Deleting topic {}", &topic_name);

    publisher
        .raw_api_mut()
        .delete_topic({
            let mut d = pubsub::api::DeleteTopicRequest::default();
            d.topic = topic_name.into();
            d
        })
        .await?;

    println!("Done");

    Ok(())
}