pipekafka 0.2.3

A pipebase plugin using rdkafka
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

use crate::constants::*;
use pipebase::common::Period;
use rdkafka::{
    client::ClientContext,
    config::{ClientConfig, FromClientConfigAndContext, RDKafkaLogLevel},
    error::KafkaResult,
};
use std::collections::HashMap;

use serde::Deserialize;

/// [reference](https://docs.confluent.io/3.2.1/clients/librdkafka/CONFIGURATION_8md.html)
/// Kafka client base config for consumer and producer
#[derive(Clone, Deserialize)]
pub struct KafkaClientConfig {
    bootstrap_servers: String,
}

impl From<KafkaClientConfig> for HashMap<&str, String> {
    fn from(config: KafkaClientConfig) -> Self {
        let mut params = HashMap::new();
        params.insert(BOOTSTRAP_SERVERS, config.bootstrap_servers);
        params
    }
}

/// Kafka client config for consumer
#[derive(Clone, Deserialize)]
pub struct KafkaConsumerClientConfig {
    group_id: Option<String>,
    enable_partition_eof: Option<bool>,
    session_timeout_ms: Option<u32>,
    enable_auto_commit: Option<bool>,
    topics: Vec<String>,
}

impl KafkaConsumerClientConfig {
    pub fn get_topics(&self) -> Vec<&str> {
        self.topics.iter().map(|t| t.as_str()).collect()
    }
}

impl From<KafkaConsumerClientConfig> for HashMap<&str, String> {
    fn from(config: KafkaConsumerClientConfig) -> Self {
        let mut params: HashMap<&str, String> = HashMap::new();
        match config.group_id {
            Some(ref group_id) => params.insert(GROUP_ID, group_id.to_owned()),
            None => None,
        };
        match config.enable_partition_eof {
            Some(ref enable_partition_eof) => {
                params.insert(ENABLE_PARTITION_EOF, format!("{}", enable_partition_eof))
            }
            None => None,
        };
        match config.session_timeout_ms {
            Some(ref session_timeout_ms) => {
                params.insert(SESSION_TIMEOUT_MS, format!("{}", session_timeout_ms))
            }
            None => None,
        };
        match config.enable_auto_commit {
            Some(ref enable_auto_commit) => {
                params.insert(ENABLE_AUTO_COMMIT, format!("{}", enable_auto_commit))
            }
            None => None,
        };
        params
    }
}

/// Kafka client config for producer
#[derive(Clone, Deserialize)]
pub struct KafkaProducerClientConfig {
    topic: String,
    queue_timeout: Period,
    message_timeout_ms: Option<u32>,
}

impl KafkaProducerClientConfig {
    pub fn get_topic(&self) -> &str {
        self.topic.as_str()
    }

    pub fn get_queue_timeout(&self) -> Period {
        self.queue_timeout.to_owned()
    }
}

impl From<KafkaProducerClientConfig> for HashMap<&str, String> {
    fn from(config: KafkaProducerClientConfig) -> Self {
        let mut params: HashMap<&str, String> = HashMap::new();
        match config.message_timeout_ms {
            Some(ref message_timeout_ms) => {
                params.insert(MESSAGE_TIMEOUT_MS, format!("{}", message_timeout_ms))
            }
            None => None,
        };
        params
    }
}

/// Create kafka consumer or producer with context
pub fn create_kafka_client<C, T>(paras: HashMap<&str, String>, context: C) -> KafkaResult<T>
where
    C: ClientContext,
    T: FromClientConfigAndContext<C>,
{
    let mut config = ClientConfig::new();
    for (prop, value) in paras {
        config.set(prop, value);
    }
    config.set_log_level(RDKafkaLogLevel::Error);
    config.create_with_context::<C, T>(context)
}