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

use rdkafka::{
    metadata::{Metadata, MetadataBroker, MetadataPartition, MetadataTopic},
    types::RDKafkaRespErr,
};
use std::convert::From;

#[derive(Debug)]
pub struct RedpandaMetadata {
    /// ID of the broker originating this metadata
    pub orig_broker_id: i32,
    /// hostname of the broker originating this metadata
    pub orig_broker_name: String,
    /// metadata information for all the brokers in the cluster
    pub brokers: Vec<RedpandaBroker>,
    /// metadata information for all the topics in the cluster
    pub topics: Vec<RedpandaTopic>,
}

impl RedpandaMetadata {
    /// Get a sorted list of the names of all topics in the cluster
    pub fn topic_names(&self) -> Vec<String> {
        let mut topic_names = Vec::new();
        for topic in &self.topics {
            topic_names.push(topic.name.clone());
        }
        topic_names.sort();

        topic_names
    }
}

impl From<Metadata> for RedpandaMetadata {
    fn from(m: Metadata) -> Self {
        let mut brokers: Vec<RedpandaBroker> = Vec::new();
        for b in m.brokers() {
            brokers.push(b.into());
        }

        let mut topics: Vec<RedpandaTopic> = Vec::new();
        for t in m.topics() {
            topics.push(t.into());
        }
        Self {
            orig_broker_id: m.orig_broker_id(),
            orig_broker_name: m.orig_broker_name().to_owned(),
            brokers,
            topics,
        }
    }
}

#[derive(Debug)]
pub struct RedpandaBroker {
    /// ID of broker
    pub id: i32,
    /// hostname of broker
    pub hostname: String,
    /// port of broker
    pub port: u16,
}

impl From<&MetadataBroker> for RedpandaBroker {
    fn from(metadata_broker: &MetadataBroker) -> Self {
        let port = metadata_broker
            .port()
            .try_into()
            .expect("Failed to convert port to u16; max port number is 65,535");
        Self {
            id: metadata_broker.id(),
            hostname: metadata_broker.host().to_owned(),
            port,
        }
    }
}

#[derive(Debug)]
pub struct RedpandaTopic {
    /// name of the topic
    pub name: String,
    /// partition metadata information for all partitions
    pub partitions: Vec<RedpandaPartition>,
    /// metadata error for the topic, or `None` if there was no error
    pub error: Option<RDKafkaRespErr>,
}

impl From<&MetadataTopic> for RedpandaTopic {
    fn from(metadata_topic: &MetadataTopic) -> Self {
        let mut partitions: Vec<RedpandaPartition> = Vec::new();
        for p in metadata_topic.partitions() {
            partitions.push(p.into());
        }

        Self {
            name: metadata_topic.name().to_owned(),
            partitions,
            error: metadata_topic.error(),
        }
    }
}

#[derive(Debug)]
pub struct RedpandaPartition {
    /// id of the partition
    pub id: i32,
    /// broker id of the leader broker for the partition
    pub leader: i32,
    /// metadata error for the partition, or `None` if there is no error
    pub error: Option<RDKafkaRespErr>,
    /// broker id of replicas
    pub replicas: Vec<i32>,
    /// broker IDs of the in-sync replicas
    pub in_sync_replicas: Vec<i32>,
}

/// Construct RedpandaPartition from rdkafka::metadata::MetadataPartition
impl From<&MetadataPartition> for RedpandaPartition {
    fn from(metadata_partition: &MetadataPartition) -> Self {
        Self {
            id: metadata_partition.id(),
            leader: metadata_partition.leader(),
            error: metadata_partition.error(),
            replicas: metadata_partition.replicas().to_owned(),
            in_sync_replicas: metadata_partition.isr().to_owned(),
        }
    }
}