redpanda 0.5.0

A crate for interacting with Redpanda Message Brokers
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

use rdkafka::{
    consumer::{MessageStream, StreamConsumer},
    error::KafkaError,
    message::BorrowedMessage,
    util::Timeout,
};
use tracing::{event, instrument, Level};

use crate::metadata::RedpandaMetadata;

pub use rdkafka::consumer::Consumer;
pub use rdkafka::consumer::CommitMode;

pub struct RedpandaConsumer {
    pub consumer: StreamConsumer,
    request_timeout: Timeout,
}

impl RedpandaConsumer {
    /// Create a new RedpandaConsumer, validating that the brokers respond to connections within timeout
    #[instrument(skip(consumer))]
    pub fn new(consumer: StreamConsumer, request_timeout: Timeout) -> Result<Self, KafkaError> {
        match consumer.fetch_metadata(Option::None, request_timeout) {
            Ok(m) => {
                let m: RedpandaMetadata = m.into();
                event!(
                    Level::INFO,
                    "Connected consumer to Redpanda cluster {:?}",
                    m
                );
                m
            }
            Err(e) => return Err(e),
        };

        Ok(Self {
            consumer,
            request_timeout,
        })
    }

    /// Get consumer metadata
    pub fn fetch_metadata(&self) -> Result<RedpandaMetadata, KafkaError> {
        let metadata = self
            .consumer
            .fetch_metadata(Option::None, self.request_timeout)?
            .into();

        Ok(metadata)
    }

    /// Subscribe the consumer to an array of topic names, checking that the topic names are valid
    /// 
    /// Subsequent calls will replace existing topics and only subscribe to the new topics provided
    #[instrument(skip(self))]
    pub fn subscribe(&self, topic_names: &[&str]) -> Result<(), KafkaError> {
        let cluster_topic_names = self.fetch_metadata()?.topic_names();

        for topic in topic_names {
            let valid_name = cluster_topic_names.binary_search(&topic.to_owned().to_owned());
            if valid_name.is_err() {
                let e = KafkaError::Subscription(format!("Invalid topic name {}", topic));
                return Err(e);
            }
        }

        match self.consumer.subscribe(topic_names) {
            Ok(_) => {
                event!(Level::INFO, "Subscribed to topics {:?}", topic_names);
                Ok(())
            }
            Err(e) => Err(e),
        }
    }

    /// Get the names of the currently subscribed topics
    pub fn get_subscription_topic_names(&self) -> Vec<String> {
        let topic_partition_list = self.consumer.subscription().unwrap();
        let mut topic_names = Vec::new();
        for elem in topic_partition_list.elements() {
            topic_names.push(elem.topic().to_owned());
        }

        topic_names
    }

    /// Receive a single message
    pub async fn recv(&self) -> Result<BorrowedMessage<'_>, KafkaError> {
        self.consumer.recv().await
    }

    /// Create a message stream from the subscribed topics
    pub fn stream(&self) -> MessageStream {
        self.consumer.stream()
    }
}