dexther_lib 0.0.261

Dexther tools for the platform.
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

pub mod types;
pub mod compression;
pub mod mongo {
    use eyre::Result;
    use mongodb::bson::Document;
    use mongodb::error::{ErrorKind, WriteFailure};
    use mongodb::{Client as MongoClient, Collection};
    use std::env;

    pub async fn connect_mongo() -> Result<MongoClient> {
        let mongo_uri = env::var("MONGO_URI")?;
        let client = MongoClient::with_uri_str(&mongo_uri).await?;
        return Ok(client);
    }

    pub async fn write_to_collection(
        message: String,
        collection: &Collection<Document>,
    ) -> Result<()> {
        println!("Writing to collection: {:?}", message);
        let json_value = serde_json::from_str(&message).unwrap();
        let document = Document::from(json_value);
        let _ = match collection.insert_one(document, None).await {
            Ok(_) => {
                println!("Successfully written to collection!");
                Ok(())
            }
            Err(e) => match e.kind.as_ref() {
                ErrorKind::Write(WriteFailure::WriteError(write_error))
                    if write_error.code == 11000 =>
                {
                    println!("Duplicate key error: {:?}", e);
                    // Handle duplicate key error, e.g., by updating the existing document or ignoring the insert
                    Err(eyre::Report::new(e))
                }
                _ => {
                    println!("Error writing to collection: {:?}", e);
                    Err(eyre::Report::new(e))
                }
            },
        };
        return Ok(());
    }
}

pub mod gcppubsub {
    use chrono::{Utc, DateTime};
    use eyre::Result;
    use google_cloud_pubsub::client::google_cloud_auth::credentials::CredentialsFile;
    use google_cloud_googleapis::pubsub::v1::PubsubMessage;
    use google_cloud_pubsub::client::{Client, ClientConfig};
    use google_cloud_pubsub::publisher::Publisher;
    use google_cloud_pubsub::subscription::Subscription;

    use std::env;

    pub async fn connect_pub_sub() -> Result<Client> {
        let service_account_json = env::var("SERVICE_ACCOUNT_JSON")?;

        let credentials = CredentialsFile::new_from_str(&service_account_json).await?;
        let config = ClientConfig::default()
            .with_credentials(credentials)
            .await
            .unwrap();
        let client = Client::new(config).await.unwrap();
        return Ok(client);
    }

    pub async fn get_subscription(client: Client, sub_name: &str) -> Result<Subscription> {
        let subscription = client.subscription(&sub_name);
        return Ok(subscription);
    }

    pub async fn get_publisher(client: Client, topic_name: &str) -> Result<Publisher> {
        let topic: google_cloud_pubsub::topic::Topic = client.topic(topic_name);
        if !topic.exists(None).await? {
            panic!("Topic does not exist")
        }
        let publisher = topic.new_publisher(None);
        return Ok(publisher);
    }

    pub async fn publish_message(publisher: &Publisher, message: String, ordering_key: Option<String>, chain_id: Option<String>) -> Result<()> {
        //println!("Publishing message: {:?}", message);
        let publisher = publisher.clone();
        let task = tokio::spawn(async move {
            let mut attributes = std::collections::HashMap::new();
            
            // Add chain_id as an attribute if provided
            if let Some(chain_id) = chain_id {
                attributes.insert("chain_id".to_string(), chain_id);
            }
            
            let msg = PubsubMessage {
                data: message.into_bytes(),
                ordering_key: ordering_key.unwrap_or_default(),
                attributes,
                // Set ordering_key if needed (https://cloud.google.com/pubsub/docs/ordering)
                ..Default::default()
            };

            // Send a message. There are also `publish_bulk` and `publish_immediately` methods.
            let mut awaiter = publisher.publish(msg).await;

            // The get method blocks until a server-generated ID or an error is returned for the published message.
            awaiter.get().await
        });
        let message_id = task.await.unwrap()?;
        println!("Published message with ID: {:?} at time {:?}", message_id, Utc::now());
        Ok(())
    }
}

pub mod blockchain;