kafkaesque 0.0.15

Kafka implementations in Rust (Unofficial)
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

use crate::formats::{
    codec::{FixedLength, Read, Write},
    Result,
};
use derive_more::{From, Into};
use tokio::io::{AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tracing::trace;

use super::VarInt;

impl<'a> Write for &'a str {
    fn calculate_size(&self) -> i32 {
        i16::SIZE + self.len() as i32
    }
    async fn write_to(&self, writer: &mut (dyn AsyncWrite + Send + Unpin)) -> Result<()> {
        let len = self.len() as i16;
        len.write_to(writer).await?;
        writer.write_all(self.as_bytes()).await?;
        Ok(())
    }
}

impl Write for String {
    fn calculate_size(&self) -> i32 {
        self.as_str().calculate_size()
    }
    async fn write_to(&self, writer: &mut (dyn AsyncWrite + Send + Unpin)) -> Result<()> {
        self.as_str().write_to(writer).await
    }
}

impl Read for String {
    async fn read_from(reader: &mut (dyn tokio::io::AsyncRead + Send + Unpin)) -> Result<Self> {
        trace!("reading string len");
        let len = i16::read_from(reader).await?;
        let mut buf = vec![0u8; len as usize];
        trace!("reading a string of len {len}");
        reader.read_exact(&mut buf).await?;
        Ok(String::from_utf8(buf)?)
    }
}

#[derive(Debug, Clone, From, Into)]
pub struct NullableString(String);

impl Write for NullableString {
    fn calculate_size(&self) -> i32 {
        i16::SIZE + self.0.len() as i32
    }
    async fn write_to(&self, writer: &mut (dyn AsyncWrite + Send + Unpin)) -> Result<()> {
        if self.0.is_empty() {
            (-1i16).write_to(writer).await
        } else {
            (self.0.len() as i16).write_to(writer).await?;
            writer.write_all(self.0.as_bytes()).await?;
            Ok(())
        }
    }
}

impl Read for NullableString {
    async fn read_from(reader: &mut (dyn tokio::io::AsyncRead + Send + Unpin)) -> Result<Self> {
        trace!("reading nullable_string len");
        let len = i16::read_from(reader).await?.max(0);
        let mut buf = vec![0u8; len as usize];
        trace!("reading a nullable_string of len {len}");
        reader.read_exact(&mut buf).await?;
        Ok(String::from_utf8(buf)?.into())
    }
}

/// A String whos length is encoded as a VarInt
#[derive(Debug, Clone, From, Into)]
pub struct VarIntString(String);

impl Write for VarIntString {
    fn calculate_size(&self) -> i32 {
        VarInt::from(self.0.len() as i64).calculate_size() + self.0.len() as i32
    }
    async fn write_to(&self, writer: &mut (dyn AsyncWrite + Send + Unpin)) -> Result<()> {
        VarInt::from(self.0.len() as i64).write_to(writer).await?;
        writer.write_all(self.0.as_bytes()).await?;
        Ok(())
    }
}

impl Read for VarIntString {
    async fn read_from(reader: &mut (dyn tokio::io::AsyncRead + Send + Unpin)) -> Result<Self> {
        let len = VarInt::read_from(reader).await?;
        let mut buf = vec![0u8; i64::from(len) as usize];
        reader.read_exact(&mut buf).await?;
        let s = String::from_utf8(buf)?.into();
        Ok(s)
    }
}