ene_kafka 0.2.0

Ene Kafka is an easy-to-use Rust client for Apache Kafka
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
116
117
118
119
120
121
122
123
124
125
126
127
128
129

use async_trait::async_trait;

use crate::dispatchers::EventDispatcher;
use crate::messages::kafka_message::KafkaTopic;
use crate::producers::producer::{KafkaProducer, KafkaProducerInterface};
use crate::{ConsumerImpl, ProducerImpl};

#[async_trait]
pub trait KafkaConsumerInterface<Dispatcher: EventDispatcher, InnerProducer: KafkaProducerInterface>
{
    fn new(consumer_group_id: String, bootstrap_servers: String) -> Self;
    async fn start<'a>(
        &'a self,
        dispatcher: &'a Dispatcher,
        dlq_producer: &'a KafkaProducer<InnerProducer>,
        topic: KafkaTopic,
        dlq_topic: KafkaTopic,
    );
}

#[derive(Debug, Clone)]
pub struct KafkaConsumer<
    Dispatcher: EventDispatcher,
    InnerConsumer: KafkaConsumerInterface<Dispatcher, InnerProducer> = ConsumerImpl,
    InnerProducer: KafkaProducerInterface = ProducerImpl,
> {
    topic: KafkaTopic,
    dlq_topic: KafkaTopic,
    dispatcher: Dispatcher,
    inner_consumer: InnerConsumer,
    dlq_producer: KafkaProducer<InnerProducer>,
}

impl<
        Dispatcher: EventDispatcher,
        Consumer: KafkaConsumerInterface<Dispatcher, InnerProducer>,
        InnerProducer: KafkaProducerInterface,
    > KafkaConsumer<Dispatcher, Consumer, InnerProducer>
{
    pub fn new(
        topic: KafkaTopic,
        dlq_topic: KafkaTopic,
        consumer_group_id: String,
        bootstrap_servers: String,
        handler: Dispatcher,
    ) -> Self {
        let dlq_producer = KafkaProducer::new(bootstrap_servers.clone());
        Self {
            topic,
            dlq_topic,
            dispatcher: handler,
            inner_consumer: Consumer::new(consumer_group_id, bootstrap_servers),
            dlq_producer,
        }
    }

    /// Starts the consumer loop
    /// This function will block the current thread
    /// It will consume messages from the Kafka topic and dispatch them to the handlers.
    /// If the message could not be consumed, it will be sent to the dead letter queue.
    pub async fn start(self) {
        self.inner_consumer
            .start(
                &self.dispatcher,
                &self.dlq_producer,
                self.topic,
                self.dlq_topic,
            )
            .await;
    }
}

///
/// Create a new Kafka consumer
/// Arguments:
/// - `topic` - a string representing the Kafka topic
/// - `dlq_topic` - a string representing the Kafka dead letter queue topic. If an event could not be consuler, it will be sent to the dead letter queue.
/// - `consumer_group_id` - a string representing the Kafka consumer group id
/// - `bootstrap_servers` - a string representing the Kafka bootstrap servers
/// - `handlers` - a list of handle declarations that will be used by this consumer
/// The handlers need to implement The `EventHandler` trait.
///
/// Example:
/// ```rust,ignore
///    let consumer = kafka_consumer!(
///        topic = KafkaTopic {
///            name: "test".to_string(),
///            content_type: ContentType::Json
///        },
///        dlq_topic = KafkaTopic {
///            name: "test-dlq".to_string(),
///            content_type: ContentType::Json
///        },
///        consumer_group_id = "test-group",
///        bootstrap_servers = bootstrap_servers,
///        handlers = {
///            entity_created_event_handler: EntityCreatedEventHandler = EntityCreatedEventHandler {}
///        }
///    );
/// consumer.start().await;
/// ```
///
#[macro_export]
macro_rules! kafka_consumer {
    (
        topic = $topic: expr,
        dlq_topic = $dlq_topic: expr,
        consumer_group_id = $consumer_group_id: expr,
        bootstrap_servers = $bootstrap_servers: expr,
        handlers = {$($handler_name: ident: $handler_type: ident = $handler: expr),*}$(,)?
        $(,)?
    ) => {
        {

            ene_kafka::generate_event_dispatcher!($($handler_name: $handler_type),*);


            ene_kafka::consumers::consumer::KafkaConsumer::<CloudEventDispatcher>::new(
                $topic,
                $dlq_topic,
                $consumer_group_id.to_string(),
                $bootstrap_servers.to_string(),
                CloudEventDispatcher { $($handler_name: $handler),* }
            )

        }

    };
}