rdkafka 0.8.1

Rust wrapper for librdkafka
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

//! Metadata fetch API
extern crate rdkafka_sys as rdkafka;

use std::ffi::CStr;
use std::os::raw::c_void;
use std::ptr;
use std::marker::PhantomData;

use log::LogLevel;

use self::rdkafka::types::*;

use config::TopicConfig;
use error::{IsError, KafkaResult, KafkaError};
use util::{bytes_cstr_to_owned, cstr_to_owned};

pub fn get_native_metadata(client: *mut RDKafka) -> KafkaResult<Metadata> {
    let mut metadata_ptr: *const RDKafkaMetadata = ptr::null_mut();
    let ret = unsafe {
        rdkafka::rd_kafka_metadata(client, 1, ptr::null::<u8>() as *mut RDKafkaTopic, &mut metadata_ptr as *mut *const RDKafkaMetadata, 2000)
    };
    if ret.is_error() {
        return Err(KafkaError::MetadataFetch(ret));
    }

    Ok(Metadata(metadata_ptr))
}

#[derive(Debug)]
pub struct Metadata(*const RDKafkaMetadata);

impl Metadata {
    pub fn topic_iterator(&self) -> MetadataTopicIterator {
        MetadataTopicIterator {
            metadata: self,
            index: 0,
        }
    }

    pub fn topics(&'a self) -> &'a [RDKafkaMetadataTopic] {
        unsafe { slice::from_raw_parts(self.0.topics, self.0.topic_cnt as usize) };
    }
}

impl Drop for Metadata {
    fn drop(&mut self) {
        unsafe { rdkafka::rd_kafka_metadata_destroy(self.0) };
    }
}

#[derive(Debug)]
pub struct MetadataTopicIterator<'a> {
    metadata: &'a Metadata,
    index: isize,
}

impl<'a> Iterator for MetadataTopicIterator<'a> {
    type Item = MetadataTopic<'a>;

    fn next(&mut self) -> Option<MetadataTopic<'a>> {
        let count = unsafe { (*self.metadata.0).topic_cnt as isize };
        if self.index >= count {
            return None
        }
        let topic_info = MetadataTopic {
            ptr: unsafe { (*self.metadata.0).topics.offset(self.index) },
            _p: PhantomData
        };
        self.index += 1;

        Some(topic_info)
    }
}

#[derive(Debug)]
pub struct MetadataTopic<'a> {
    ptr: *mut RDKafkaMetadataTopic,
    _p: PhantomData<&'a u8>,
}

impl<'a> MetadataTopic<'a> {
    pub fn name(&self) -> &'a str {
        unsafe {
            CStr::from_ptr((*self.ptr).topic)
                .to_str()
                .expect("Topic name is not a valid UTF-8 string")
        }
    }

    pub fn partition_iterator(&self) -> MetadataPartitionIterator<'a> {
        MetadataPartitionIterator {
            partitions: unsafe { (*self.ptr).partitions },
            count: unsafe { (*self.ptr).partition_cnt } as isize,
            index: 0,
            _p: PhantomData,
        }
    }
}

pub struct MetadataPartitionIterator<'a> {
    partitions: *mut RDKafkaMetadataPartition,
    count: isize,
    index: isize,
    _p: PhantomData<&'a u8>
}

impl<'a> Iterator for MetadataPartitionIterator<'a> {
    type Item = MetadataPartition<'a>;

    fn next(&mut self) -> Option<MetadataPartition<'a>> {
        if self.index >= self.count {
            return None
        }
        let partition_metadata_ptr = unsafe { self.partitions.offset(self.index) };
        self.index += 1;
        Some(MetadataPartition{ptr: partition_metadata_ptr, _p: PhantomData})
    }
}

pub struct MetadataPartition<'a> {
    ptr: *mut RDKafkaMetadataPartition,
    _p: PhantomData<&'a u8>
}

impl<'a> MetadataPartition<'a> {
    pub fn id(&self) -> i32 {
        unsafe { (*self.ptr).id }
    }
}