kinbox 0.1.3

A simple kafka inbox
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

use crate::config::{CommonConfig, WriteConfig};
use rdkafka::error::KafkaError;
use rdkafka::message::ToBytes;
use rdkafka::producer::FutureRecord;
use rdkafka::util::Timeout;
use std::time::Duration;

/// 写邮箱
#[derive(Clone)]
pub struct Writer {
    producer: rdkafka::producer::FutureProducer,
}

impl Writer {
    pub fn new(cfg: &WriteConfig) -> Result<Self, KafkaError> {
        let producer: rdkafka::producer::FutureProducer = cfg.inner.create()?;
        Ok(Writer { producer })
    }

    pub fn new_with_brokers<V: AsRef<str>>(brokers: &[V]) -> Result<Self, KafkaError> {
        let mut c = WriteConfig::default();
        c.set_brokers(brokers);
        Writer::new(&c)
    }

    /// 写邮件
    pub async fn write<K, P>(&self, record: FutureRecord<'_, K, P>) -> Result<(), KafkaError>
    where
        K: ToBytes + ?Sized,
        P: ToBytes + ?Sized,
    {
        self.producer
            .send(record, Duration::from_secs(3))
            .await
            .map(|_| ())
            .map_err(|e| e.0)
    }

    /// 写邮件带超时
    pub async fn write_timeout<K, P, T>(
        &self,
        record: FutureRecord<'_, K, P>,
        queue_timeout: T,
    ) -> Result<(), KafkaError>
    where
        K: ToBytes + ?Sized,
        P: ToBytes + ?Sized,
        T: Into<Timeout>,
    {
        self.producer
            .send(record, queue_timeout)
            .await
            .map(|_| ())
            .map_err(|e| e.0)
    }
}