samsa 0.1.8

Rust-native Kafka/Redpanda protocol and client implementation.
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

//! Encoding and creation for Leave Group requests.
//!
//! To explicitly leave a group, the client can send a leave group request. This is preferred over letting the session timeout expire since it allows the group to rebalance faster, which for the consumer means that less time will elapse before partitions can be reassigned to an active member.
//!
//! ### Example
//! ```rust
//! let leave = protocol::LeaveGroupRequest::new(
//!     CORRELATION_ID,
//!     CLIENT_ID,
//!     group_id,
//!     member_id,
//! );
//! conn.send_request(&leave).await?;
//! ```
//!
//! ### Protocol Def
//! ```text
//! LeaveGroup Request (Version: 0) => group_id member_id
//!   group_id => STRING
//!   member_id => STRING
//! ```
//!
//! Note we are using version 0 for this request.

use bytes::Bytes;
use nom::AsBytes;

use crate::{
    encode::ToByte,
    error::{Error, Result},
    protocol::HeaderRequest,
};

const API_KEY_METADATA: i16 = 13;
const API_VERSION: i16 = 0;

/// The base Leave Group request object.
///
/// ### Example
/// ```rust
/// let leave = protocol::LeaveGroupRequest::new(
///     CORRELATION_ID,
///     CLIENT_ID,
///     group_id,
///     member_id,
/// );
/// conn.send_request(&leave).await?;
/// ```
#[derive(Debug)]
pub struct LeaveGroupRequest<'a> {
    pub header: HeaderRequest<'a>,
    /// The ID of the group to leave.
    pub group_id: &'a str,
    /// The member ID to remove from the group.
    pub member_id: String,
}

impl<'a> LeaveGroupRequest<'a> {
    pub fn new(
        correlation_id: i32,
        client_id: &'a str,
        group_id: &'a str,
        member_id: Bytes,
    ) -> Result<Self> {
        let header = HeaderRequest::new(API_KEY_METADATA, API_VERSION, correlation_id, client_id);
        Ok(Self {
            header,
            group_id,
            member_id: String::from_utf8(member_id.as_bytes().to_vec())
                .map_err(|_| Error::DecodingUtf8Error)?,
        })
    }
}

impl ToByte for LeaveGroupRequest<'_> {
    fn encode<T: bytes::BufMut>(&self, buffer: &mut T) -> crate::error::Result<()> {
        tracing::trace!("Encoding LeaveGroupRequest {:?}", self);
        self.header.encode(buffer)?;
        self.group_id.encode(buffer)?;
        self.member_id.encode(buffer)?;
        Ok(())
    }
}