webbase 0.0.18

webbase
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

use anyhow::anyhow;
use rdkafka::config::ClientConfig;
use rdkafka::message::{Header, OwnedHeaders};
use rdkafka::producer::{FutureProducer, FutureRecord};
use std::sync::Arc;
use std::time::Duration;

#[derive(Clone)]
pub struct KafkaClient {
    pub producer: Arc<FutureProducer>,
}

impl KafkaClient {
    pub fn new(brokers: &str) -> KafkaClient {
        let mut creator = ClientConfig::new();
        tracing::info!("conneting to: {:?}", brokers);
        creator.set("bootstrap.servers", brokers);
        // creator.set("message.send.max.retries", "3");
        // creator.set("message.timeout.ms", "5000");
        // creator.set("queue.buffering.max.ms", "5000");
        // creator.set("queue.buffering.max.kbytes", "32768");
        // creator.set("batch.num.messages", "200");
        let producer: FutureProducer = creator.create().expect("Producer creation error");
        KafkaClient {
            producer: Arc::new(producer),
        }
    }

    pub async fn send_text(&self, topic_name: &str, data: &str, key: String) -> anyhow::Result<()> {
        let _delivery_status = self
            .producer
            .send(
                FutureRecord::to(topic_name)
                    .payload(data)
                    .key(&format!("{}", key))
                    .headers(OwnedHeaders::new().insert(Header {
                        key: "header_key",
                        value: Some("header_value"),
                    })),
                Duration::from_secs(0),
            )
            .await
            .map_err(|e| {
                return Err::<(), anyhow::Error>(anyhow!("send kafka msg with error: {:?}", e));
            })
            .unwrap();
        anyhow::Ok(())
    }

    pub async fn send_binary(
        &self,
        topic_name: &str,
        data: &Vec<u8>,
        key: String,
    ) -> anyhow::Result<()> {
        let _delivery_status = self
            .producer
            .send(
                FutureRecord::to(topic_name)
                    .payload(data)
                    .key(&format!("{}", key))
                    .headers(OwnedHeaders::new().insert(Header {
                        key: "header_key",
                        value: Some("header_value"),
                    })),
                Duration::from_secs(0),
            )
            .await
            .map_err(|e| {
                return Err::<(), anyhow::Error>(anyhow!("send kafka msg with error: {:?}", e));
            })
            .unwrap();
        anyhow::Ok(())
    }
}