rabbitmq_stream_protocol/commands/
close.rs

1/*CloseRequest => Key Version CorrelationId ClosingCode ClosingReason
2
3CloseResponse => Key Version CorrelationId ResponseCode
4Key => uint16 // 22
5Version => uint16
6CorrelationId => uint32
7ResponseCode => uint16
8*/
9use std::io::Write;
10
11use crate::{
12    codec::{Decoder, Encoder},
13    error::{DecodeError, EncodeError},
14    protocol::commands::COMMAND_CLOSE,
15    FromResponse, ResponseCode,
16};
17
18use super::Command;
19
20#[cfg(test)]
21use fake::Fake;
22
23#[cfg_attr(test, derive(fake::Dummy))]
24#[derive(PartialEq, Eq, Debug)]
25pub struct CloseRequest {
26    correlation_id: u32,
27    closing_code: ResponseCode,
28    closing_reason: String,
29}
30
31impl CloseRequest {
32    pub fn new(correlation_id: u32, closing_code: ResponseCode, closing_reason: String) -> Self {
33        Self {
34            correlation_id,
35            closing_code,
36            closing_reason,
37        }
38    }
39}
40
41impl Encoder for CloseRequest {
42    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
43        self.correlation_id.encode(writer)?;
44        self.closing_code.encode(writer)?;
45        self.closing_reason.as_str().encode(writer)?;
46        Ok(())
47    }
48
49    fn encoded_size(&self) -> u32 {
50        self.correlation_id.encoded_size()
51            + self.closing_code.encoded_size()
52            + self.closing_reason.as_str().encoded_size()
53    }
54}
55
56impl Command for CloseRequest {
57    fn key(&self) -> u16 {
58        COMMAND_CLOSE
59    }
60}
61impl Decoder for CloseRequest {
62    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
63        let (input, correlation_id) = u32::decode(input)?;
64        let (input, closing_code) = ResponseCode::decode(input)?;
65        let (input, closing_reason) = Option::decode(input)?;
66
67        Ok((
68            input,
69            CloseRequest {
70                correlation_id,
71                closing_code,
72                closing_reason: closing_reason.unwrap(),
73            },
74        ))
75    }
76}
77
78#[cfg_attr(test, derive(fake::Dummy))]
79#[derive(PartialEq, Eq, Debug)]
80pub struct CloseResponse {
81    pub(crate) correlation_id: u32,
82    response_code: ResponseCode,
83}
84
85impl CloseResponse {
86    pub fn new(correlation_id: u32, response_code: ResponseCode) -> Self {
87        Self {
88            correlation_id,
89            response_code,
90        }
91    }
92    pub fn is_ok(&self) -> bool {
93        self.response_code == ResponseCode::Ok
94    }
95}
96
97impl Encoder for CloseResponse {
98    fn encode(&self, writer: &mut impl Write) -> Result<(), EncodeError> {
99        self.correlation_id.encode(writer)?;
100        self.response_code.encode(writer)?;
101        Ok(())
102    }
103
104    fn encoded_size(&self) -> u32 {
105        self.correlation_id.encoded_size() + self.response_code.encoded_size()
106    }
107}
108
109impl Decoder for CloseResponse {
110    fn decode(input: &[u8]) -> Result<(&[u8], Self), DecodeError> {
111        let (input, correlation_id) = u32::decode(input)?;
112        let (input, response_code) = ResponseCode::decode(input)?;
113
114        Ok((
115            input,
116            CloseResponse {
117                correlation_id,
118                response_code,
119            },
120        ))
121    }
122}
123
124impl FromResponse for CloseResponse {
125    fn from_response(response: crate::Response) -> Option<Self> {
126        match response.kind {
127            crate::ResponseKind::Close(close) => Some(close),
128            _ => None,
129        }
130    }
131}
132
133#[cfg(test)]
134mod tests {
135    use super::CloseRequest;
136    use super::CloseResponse;
137    use crate::commands::tests::command_encode_decode_test;
138
139    #[test]
140    fn close_request_test() {
141        command_encode_decode_test::<CloseRequest>()
142    }
143
144    #[test]
145    fn close_response_test() {
146        command_encode_decode_test::<CloseResponse>()
147    }
148}
rabbitmq_stream_protocol/commands/close.rs

rabbitmq_stream_protocol/commands/
close.rs
