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
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

use futures_util::StreamExt;
use rdkafka::client::ClientContext;
#[allow(unused_imports)]
use rdkafka::config::{ClientConfig, RDKafkaLogLevel};
use rdkafka::consumer::stream_consumer::StreamConsumer;
#[allow(unused_imports)]
use rdkafka::consumer::{CommitMode, Consumer, ConsumerContext, Rebalance};
use rdkafka::error::KafkaResult;
use rdkafka::topic_partition_list::TopicPartitionList;
use rdkafka::Message;
use tokio::sync::mpsc::UnboundedSender;
use tracing::{info, trace, warn};

pub struct CustomContext;
impl ClientContext for CustomContext {}
impl ConsumerContext for CustomContext {
    fn pre_rebalance(&self, rebalance: &Rebalance) {
        info!("Pre rebalance {:?}", rebalance);
    }

    fn post_rebalance(&self, rebalance: &Rebalance) {
        info!("Post rebalance {:?}", rebalance);
    }

    fn commit_callback(&self, result: KafkaResult<()>, _offsets: &TopicPartitionList) {
        trace!("Committing offsets: {:?}", result);
    }
}

pub struct KafkaConsumer {}

#[derive(Debug, Clone)]
pub struct ConsumerMsg {
    pub key: Option<Vec<u8>>,
    pub payload: Option<Vec<u8>>,
    pub topic: String,
    pub timestamp: Option<i64>,
    pub partition: i32,
    pub offset: i64,
}

impl KafkaConsumer {
    pub fn start_consume(
        topic: String,
        group_id: String,
        brokers: String,
        sender: UnboundedSender<ConsumerMsg>,
    ) -> anyhow::Result<()> {
        let mut creator = ClientConfig::new();
        creator.set("group.id", group_id);
        creator.set("bootstrap.servers", brokers);
        creator.set("enable.partition.eof", "false");
        creator.set("session.timeout.ms", "6000");
        creator.set("enable.auto.commit", "false");
        // creator.set("statistics.interval.ms", "30000");
        // creator.set("auto.offset.reset", "smallest");
        // creator.set_log_level(RDKafkaLogLevel::Debug);

        let context = CustomContext;
        let consumer: StreamConsumer<CustomContext> = creator
            .create_with_context(context)
            .expect("Consumer creation failed");

        tokio::spawn(async move {
            let sender = sender;
            consumer
                .subscribe(&[topic.as_str()])
                .expect("Can't subscribe to specified topics");
            info!("Start consuming from topic: {}", topic);

            let mut message_stream = consumer.stream();
            while let Some(message) = message_stream.next().await {
                match message {
                    Err(e) => warn!("Kafka strem error: {}", e),
                    Ok(m) => {
                        // let payload = m.payload_view::<str>().unwrap();
                        // let payload = match m.payload_view::<str>() {
                        //     None => "",
                        //     Some(Ok(s)) => s,
                        //     Some(Err(e)) => {
                        //         warn!("Error while deserializing message payload: {:?}", e);
                        //         ""
                        //     }
                        // };
                        // info!("key: '{:?}', payload: '{}', topic: {}, partition: {}, offset: {}, timestamp: {:?}", m.key(), payload, m.topic(), m.partition(), m.offset(), m.timestamp());
                        // let key = String::from_utf8_lossy(m.key().unwrap()).to_string();
                        let key = if m.key().is_some() {
                            Some(m.key().unwrap().to_vec())
                        } else {
                            None
                        };
                        let payload: Option<Vec<u8>> = if m.payload().is_some() {
                            Some(m.payload().unwrap().to_vec())
                        } else {
                            None
                        };

                        let data = ConsumerMsg {
                            timestamp: m.timestamp().to_millis(),
                            topic: m.topic().to_string(),
                            partition: m.partition(),
                            key: key,
                            payload: payload,
                            offset: m.offset(),
                        };
                        let _ = sender.send(data);
                        consumer.commit_message(&m, CommitMode::Async).unwrap();
                    }
                };
            }
        });
        anyhow::Ok(())
    }
}