ntex_amqp_codec/protocol/
mod.rs

1#![allow(clippy::derivable_impls)]
2use std::fmt;
3
4use chrono::{DateTime, Utc};
5use derive_more::From;
6use ntex_bytes::{Buf, BufMut, BytePages, ByteString, Bytes};
7use uuid::Uuid;
8
9use crate::codec::{self, Decode, DecodeFormatted, Encode};
10use crate::types::{
11    Descriptor, List, Multiple, StaticSymbol, Str, Symbol, Variant, VecStringMap, VecSymbolMap,
12};
13use crate::{HashMap, error::AmqpParseError, message::Message};
14
15impl fmt::Display for Error {
16    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
17        write!(f, "{self:?}")
18    }
19}
20
21#[derive(Debug, PartialEq, Eq, Hash, Clone, Copy)]
22pub enum ProtocolId {
23    Amqp = 0,
24    AmqpTls = 2,
25    AmqpSasl = 3,
26}
27
28pub type Map = HashMap<Variant, Variant>;
29pub type StringVariantMap = HashMap<Str, Variant>;
30pub type Fields = HashMap<Symbol, Variant>;
31pub type FilterSet = HashMap<Symbol, Option<ByteString>>;
32pub type FieldsVec = VecSymbolMap;
33pub type Timestamp = DateTime<Utc>;
34pub type Symbols = Multiple<Symbol>;
35pub type IetfLanguageTags = Multiple<IetfLanguageTag>;
36pub type Annotations = HashMap<Symbol, Variant>;
37
38#[allow(
39    clippy::unreadable_literal,
40    clippy::match_bool,
41    clippy::large_enum_variant
42)]
43mod definitions;
44pub use self::definitions::*;
45
46#[derive(Debug, Eq, PartialEq, Clone, From)]
47pub enum MessageId {
48    Ulong(u64),
49    Uuid(Uuid),
50    Binary(Bytes),
51    String(ByteString),
52}
53
54impl From<usize> for MessageId {
55    fn from(id: usize) -> MessageId {
56        MessageId::Ulong(id as u64)
57    }
58}
59
60impl From<i32> for MessageId {
61    fn from(id: i32) -> MessageId {
62        MessageId::Ulong(id as u64)
63    }
64}
65
66impl DecodeFormatted for MessageId {
67    fn decode_with_format(input: &mut Bytes, fmt: u8) -> Result<Self, AmqpParseError> {
68        match fmt {
69            codec::FORMATCODE_SMALLULONG | codec::FORMATCODE_ULONG | codec::FORMATCODE_ULONG_0 => {
70                u64::decode_with_format(input, fmt).map(MessageId::Ulong)
71            }
72            codec::FORMATCODE_UUID => Uuid::decode_with_format(input, fmt).map(MessageId::Uuid),
73            codec::FORMATCODE_BINARY8 | codec::FORMATCODE_BINARY32 => {
74                Bytes::decode_with_format(input, fmt).map(MessageId::Binary)
75            }
76            codec::FORMATCODE_STRING8 | codec::FORMATCODE_STRING32 => {
77                ByteString::decode_with_format(input, fmt).map(MessageId::String)
78            }
79            _ => Err(AmqpParseError::InvalidFormatCode(fmt)),
80        }
81    }
82}
83
84impl Encode for MessageId {
85    fn encoded_size(&self) -> usize {
86        match *self {
87            MessageId::Ulong(v) => v.encoded_size(),
88            MessageId::Uuid(ref v) => v.encoded_size(),
89            MessageId::Binary(ref v) => v.encoded_size(),
90            MessageId::String(ref v) => v.encoded_size(),
91        }
92    }
93
94    fn encode(&self, buf: &mut BytePages) {
95        match *self {
96            MessageId::Ulong(v) => v.encode(buf),
97            MessageId::Uuid(ref v) => v.encode(buf),
98            MessageId::Binary(ref v) => v.encode(buf),
99            MessageId::String(ref v) => v.encode(buf),
100        }
101    }
102}
103
104#[derive(Clone, Debug, PartialEq, Eq, From)]
105pub enum ErrorCondition {
106    AmqpError(AmqpError),
107    ConnectionError(ConnectionError),
108    SessionError(SessionError),
109    LinkError(LinkError),
110    Custom(Symbol),
111}
112
113impl Default for ErrorCondition {
114    fn default() -> ErrorCondition {
115        ErrorCondition::Custom(Symbol(Str::from("Unknown")))
116    }
117}
118
119impl DecodeFormatted for ErrorCondition {
120    #[inline]
121    fn decode_with_format(input: &mut Bytes, format: u8) -> Result<Self, AmqpParseError> {
122        let result = Symbol::decode_with_format(input, format)?;
123        if let Ok(r) = AmqpError::try_from(&result) {
124            return Ok(ErrorCondition::AmqpError(r));
125        }
126        if let Ok(r) = ConnectionError::try_from(&result) {
127            return Ok(ErrorCondition::ConnectionError(r));
128        }
129        if let Ok(r) = SessionError::try_from(&result) {
130            return Ok(ErrorCondition::SessionError(r));
131        }
132        if let Ok(r) = LinkError::try_from(&result) {
133            return Ok(ErrorCondition::LinkError(r));
134        }
135        Ok(ErrorCondition::Custom(result))
136    }
137}
138
139impl Encode for ErrorCondition {
140    fn encoded_size(&self) -> usize {
141        match *self {
142            ErrorCondition::AmqpError(ref v) => v.encoded_size(),
143            ErrorCondition::ConnectionError(ref v) => v.encoded_size(),
144            ErrorCondition::SessionError(ref v) => v.encoded_size(),
145            ErrorCondition::LinkError(ref v) => v.encoded_size(),
146            ErrorCondition::Custom(ref v) => v.encoded_size(),
147        }
148    }
149
150    fn encode(&self, buf: &mut BytePages) {
151        match *self {
152            ErrorCondition::AmqpError(ref v) => v.encode(buf),
153            ErrorCondition::ConnectionError(ref v) => v.encode(buf),
154            ErrorCondition::SessionError(ref v) => v.encode(buf),
155            ErrorCondition::LinkError(ref v) => v.encode(buf),
156            ErrorCondition::Custom(ref v) => v.encode(buf),
157        }
158    }
159}
160
161#[derive(Clone, Debug, PartialEq, Eq)]
162pub enum DistributionMode {
163    Move,
164    Copy,
165    Custom(Symbol),
166}
167
168impl DecodeFormatted for DistributionMode {
169    fn decode_with_format(input: &mut Bytes, format: u8) -> Result<Self, AmqpParseError> {
170        let result = Symbol::decode_with_format(input, format)?;
171        let result = match result.as_str() {
172            "move" => DistributionMode::Move,
173            "copy" => DistributionMode::Copy,
174            _ => DistributionMode::Custom(result),
175        };
176        Ok(result)
177    }
178}
179
180impl Encode for DistributionMode {
181    fn encoded_size(&self) -> usize {
182        match *self {
183            DistributionMode::Move => 6,
184            DistributionMode::Copy => 6,
185            DistributionMode::Custom(ref v) => v.encoded_size(),
186        }
187    }
188
189    fn encode(&self, buf: &mut BytePages) {
190        match *self {
191            DistributionMode::Move => Symbol::from("move").encode(buf),
192            DistributionMode::Copy => Symbol::from("copy").encode(buf),
193            DistributionMode::Custom(ref v) => v.encode(buf),
194        }
195    }
196}
197
198impl SaslInit {
199    pub fn prepare_response(authz_id: &str, authn_id: &str, password: &str) -> Bytes {
200        Bytes::from(format!("{authz_id}\x00{authn_id}\x00{password}"))
201    }
202}
203
204#[derive(Debug, Clone, From)]
205pub enum TransferBody {
206    Data(Bytes),
207    Pages(BytePages),
208    Message(Message),
209}
210
211impl TransferBody {
212    #[inline]
213    pub fn len(&self) -> usize {
214        self.encoded_size()
215    }
216
217    #[inline]
218    pub fn message_format(&self) -> Option<MessageFormat> {
219        match self {
220            TransferBody::Data(_) | TransferBody::Pages(_) => None,
221            TransferBody::Message(data) => data.0.message_format,
222        }
223    }
224}
225
226impl Encode for TransferBody {
227    #[inline]
228    fn encoded_size(&self) -> usize {
229        match self {
230            TransferBody::Data(data) => data.len(),
231            TransferBody::Pages(data) => data.len(),
232            TransferBody::Message(data) => data.encoded_size(),
233        }
234    }
235
236    #[inline]
237    fn encode(&self, dst: &mut BytePages) {
238        match *self {
239            TransferBody::Data(ref data) => dst.append(data),
240            TransferBody::Pages(ref data) => data.copy_to(dst),
241            TransferBody::Message(ref data) => data.encode(dst),
242        }
243    }
244}
245
246impl Eq for TransferBody {}
247
248impl PartialEq for TransferBody {
249    fn eq(&self, other: &TransferBody) -> bool {
250        match self {
251            TransferBody::Data(data) => {
252                if let TransferBody::Data(d) = other {
253                    data == d
254                } else {
255                    false
256                }
257            }
258            TransferBody::Message(msg) => {
259                if let TransferBody::Message(msg2) = other {
260                    msg == msg2
261                } else {
262                    false
263                }
264            }
265            TransferBody::Pages(_) => false,
266        }
267    }
268}
269
270impl Transfer {
271    #[inline]
272    pub fn get_body(&self) -> Option<&Bytes> {
273        match self.body() {
274            Some(TransferBody::Data(b)) => Some(b),
275            _ => None,
276        }
277    }
278
279    #[inline]
280    pub fn load_message<T: Decode>(&self) -> Result<T, AmqpParseError> {
281        if let Some(TransferBody::Data(b)) = self.body() {
282            Ok(T::decode(&mut b.clone())?)
283        } else {
284            Err(AmqpParseError::UnexpectedType("body"))
285        }
286    }
287}
288
289impl Default for Role {
290    fn default() -> Role {
291        Role::Sender
292    }
293}
294
295impl Default for SenderSettleMode {
296    fn default() -> SenderSettleMode {
297        SenderSettleMode::Mixed
298    }
299}
300
301impl Default for ReceiverSettleMode {
302    fn default() -> ReceiverSettleMode {
303        ReceiverSettleMode::First
304    }
305}
306
307impl Default for TerminusDurability {
308    fn default() -> TerminusDurability {
309        TerminusDurability::None
310    }
311}
312
313impl Default for TerminusExpiryPolicy {
314    fn default() -> TerminusExpiryPolicy {
315        TerminusExpiryPolicy::LinkDetach
316    }
317}
318
319impl Default for SaslCode {
320    fn default() -> SaslCode {
321        SaslCode::Ok
322    }
323}
324
325#[cfg(test)]
326mod tests {
327    use uuid::Uuid;
328
329    use super::*;
330    use crate::codec::{Decode, Encode};
331    use crate::error::AmqpCodecError;
332
333    #[test]
334    fn test_message_id() -> Result<(), AmqpCodecError> {
335        let id = MessageId::Uuid(Uuid::new_v4());
336
337        let mut buf = BytePages::default();
338        id.encode(&mut buf);
339
340        let new_id = MessageId::decode(&mut buf.freeze())?;
341        assert_eq!(id, new_id);
342        Ok(())
343    }
344
345    #[test]
346    fn test_properties() -> Result<(), AmqpCodecError> {
347        let id = Uuid::new_v4();
348        let props = Properties {
349            correlation_id: Some(id.into()),
350            ..Default::default()
351        };
352
353        let mut buf = BytePages::default();
354        props.encode(&mut buf);
355
356        let props2 = Properties::decode(&mut buf.freeze())?;
357        assert_eq!(props, props2);
358        Ok(())
359    }
360}
ntex_amqp_codec/protocol/mod.rs

ntex_amqp_codec/protocol/
mod.rs
