kacrab 0.2.0

A Kafka client for Rust, built from the protocol up.
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

//! Typed value deserializers, the consumer analogue of the producer's
//! `ProducerSerializer`. `poll` returns bytes-first records; a deserializer maps
//! a record's raw `key`/`value` bytes into a typed value on demand.

use bytes::Bytes;

use super::error::{ConsumerError, Result};

/// Converts raw Kafka record bytes into a typed key or value.
pub trait ConsumerDeserializer<T>: Send + Sync + 'static {
    /// Deserialize one key or value. `None` bytes (a null key/tombstone value)
    /// deserialize to `None`.
    ///
    /// # Errors
    /// Returns [`ConsumerError::Deserialization`] when the bytes cannot be
    /// decoded into `T`.
    fn deserialize(&self, topic: &str, bytes: Option<&Bytes>) -> Result<Option<T>>;
}

/// Identity deserializer: returns the raw [`Bytes`] unchanged.
#[derive(Debug, Clone, Copy, Default)]
pub struct BytesDeserializer;

impl ConsumerDeserializer<Bytes> for BytesDeserializer {
    fn deserialize(&self, _topic: &str, bytes: Option<&Bytes>) -> Result<Option<Bytes>> {
        Ok(bytes.cloned())
    }
}

/// Deserializes record bytes into an owned byte vector.
#[derive(Debug, Clone, Copy, Default)]
pub struct ByteArrayDeserializer;

impl ConsumerDeserializer<Vec<u8>> for ByteArrayDeserializer {
    fn deserialize(&self, _topic: &str, bytes: Option<&Bytes>) -> Result<Option<Vec<u8>>> {
        Ok(bytes.map(|bytes| bytes.to_vec()))
    }
}

/// Deserializes UTF-8 record bytes into a [`String`], mirroring Kafka's
/// `StringDeserializer`.
#[derive(Debug, Clone, Copy, Default)]
pub struct StringDeserializer;

impl ConsumerDeserializer<String> for StringDeserializer {
    fn deserialize(&self, _topic: &str, bytes: Option<&Bytes>) -> Result<Option<String>> {
        bytes
            .map(|bytes| {
                std::str::from_utf8(bytes)
                    .map(str::to_owned)
                    .map_err(|_error| {
                        ConsumerError::Deserialization("record bytes are not valid UTF-8")
                    })
            })
            .transpose()
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[test]
    fn string_deserializer_decodes_utf8_and_null() {
        let de = StringDeserializer;
        assert_eq!(
            de.deserialize("t", Some(&Bytes::from_static(b"hi")))
                .unwrap(),
            Some("hi".to_owned())
        );
        assert_eq!(de.deserialize("t", None).unwrap(), None);
        assert!(
            de.deserialize("t", Some(&Bytes::from_static(&[0xff, 0xfe])))
                .is_err()
        );
    }

    #[test]
    fn byte_deserializers_pass_bytes_through() {
        let raw = Bytes::from_static(b"abc");
        assert_eq!(
            BytesDeserializer.deserialize("t", Some(&raw)).unwrap(),
            Some(raw.clone())
        );
        assert_eq!(
            ByteArrayDeserializer.deserialize("t", Some(&raw)).unwrap(),
            Some(b"abc".to_vec())
        );
    }
}