use bytes::Bytes;
use super::decode::{
Decode, DecodeError, peek_code, read_array_header, read_bytes, read_list_header, read_u32,
read_u8,
};
use super::primitives::{Symbol, codes};
#[derive(Debug, Clone, PartialEq, Eq)]
pub enum Descriptor {
Code(u64),
Symbol(Symbol),
}
impl std::fmt::Display for Descriptor {
fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
match self {
Descriptor::Code(c) => write!(f, "{c:#x}"),
Descriptor::Symbol(s) => write!(f, "{s}"),
}
}
}
#[allow(missing_docs)]
pub mod descriptors {
pub const OPEN: u64 = 0x0000_0010;
pub const BEGIN: u64 = 0x0000_0011;
pub const ATTACH: u64 = 0x0000_0012;
pub const FLOW: u64 = 0x0000_0013;
pub const TRANSFER: u64 = 0x0000_0014;
pub const DISPOSITION: u64 = 0x0000_0015;
pub const DETACH: u64 = 0x0000_0016;
pub const END: u64 = 0x0000_0017;
pub const CLOSE: u64 = 0x0000_0018;
pub const ERROR: u64 = 0x0000_001d;
pub const COORDINATOR: u64 = 0x0000_0030;
pub const DECLARE: u64 = 0x0000_0031;
pub const DISCHARGE: u64 = 0x0000_0032;
pub const DECLARED: u64 = 0x0000_0033;
pub const TRANSACTIONAL_STATE: u64 = 0x0000_0034;
pub const RECEIVED: u64 = 0x0000_0023;
pub const ACCEPTED: u64 = 0x0000_0024;
pub const REJECTED: u64 = 0x0000_0025;
pub const RELEASED: u64 = 0x0000_0026;
pub const MODIFIED: u64 = 0x0000_0027;
pub const SOURCE: u64 = 0x0000_0028;
pub const TARGET: u64 = 0x0000_0029;
pub const DELETE_ON_CLOSE: u64 = 0x0000_002b;
pub const DELETE_ON_NO_LINKS: u64 = 0x0000_002c;
pub const DELETE_ON_NO_MESSAGES: u64 = 0x0000_002d;
pub const DELETE_ON_NO_LINKS_OR_MESSAGES: u64 = 0x0000_002e;
pub const HEADER: u64 = 0x0000_0070;
pub const DELIVERY_ANNOTATIONS: u64 = 0x0000_0071;
pub const MESSAGE_ANNOTATIONS: u64 = 0x0000_0072;
pub const PROPERTIES: u64 = 0x0000_0073;
pub const APPLICATION_PROPERTIES: u64 = 0x0000_0074;
pub const DATA: u64 = 0x0000_0075;
pub const AMQP_SEQUENCE: u64 = 0x0000_0076;
pub const AMQP_VALUE: u64 = 0x0000_0077;
pub const FOOTER: u64 = 0x0000_0078;
pub const SASL_MECHANISMS: u64 = 0x0000_0040;
pub const SASL_INIT: u64 = 0x0000_0041;
pub const SASL_CHALLENGE: u64 = 0x0000_0042;
pub const SASL_RESPONSE: u64 = 0x0000_0043;
pub const SASL_OUTCOME: u64 = 0x0000_0044;
}
pub fn decode_descriptor(buf: &mut Bytes) -> Result<Descriptor, DecodeError> {
match peek_code(buf) {
Some(codes::ULONG_0 | codes::SMALL_ULONG | codes::ULONG) => {
Ok(Descriptor::Code(u64::decode(buf)?))
}
Some(codes::SYM8 | codes::SYM32) => Ok(Descriptor::Symbol(Symbol::decode(buf)?)),
Some(c) => Err(DecodeError::InvalidFormatCode {
code: c,
expected: "descriptor (ulong or symbol)",
}),
None => Err(DecodeError::Eof { needed: 1 }),
}
}
pub fn peek_descriptor(buf: &Bytes) -> Result<Descriptor, DecodeError> {
let mut probe = buf.clone();
match read_u8(&mut probe)? {
codes::DESCRIBED => decode_descriptor(&mut probe),
c => Err(DecodeError::InvalidFormatCode {
code: c,
expected: "described type (0x00)",
}),
}
}
pub fn decode_described_list(buf: &mut Bytes, expected: u64) -> Result<ListDecoder, DecodeError> {
match read_u8(buf)? {
codes::DESCRIBED => {}
c => {
return Err(DecodeError::InvalidFormatCode {
code: c,
expected: "described type (0x00)",
});
}
}
match decode_descriptor(buf)? {
Descriptor::Code(c) if c == expected => {}
other => {
return Err(DecodeError::UnexpectedDescriptor {
expected,
found: other.to_string(),
});
}
}
let (count, body) = read_list_header(buf)?;
Ok(ListDecoder {
body,
remaining: count,
})
}
pub fn expect_descriptor(buf: &mut Bytes, expected: u64) -> Result<(), DecodeError> {
match read_u8(buf)? {
codes::DESCRIBED => {}
c => {
return Err(DecodeError::InvalidFormatCode {
code: c,
expected: "described type (0x00)",
});
}
}
match decode_descriptor(buf)? {
Descriptor::Code(c) if c == expected => Ok(()),
other => Err(DecodeError::UnexpectedDescriptor {
expected,
found: other.to_string(),
}),
}
}
#[derive(Debug)]
pub struct ListDecoder {
body: Bytes,
remaining: u32,
}
impl ListDecoder {
pub fn remaining(&self) -> u32 {
self.remaining
}
fn advance_one(&mut self) {
self.remaining -= 1;
}
pub fn opt<T: Decode>(&mut self) -> Result<Option<T>, DecodeError> {
if self.remaining == 0 {
return Ok(None);
}
self.advance_one();
if peek_code(&self.body) == Some(codes::NULL) {
let _ = read_u8(&mut self.body)?;
Ok(None)
} else {
Ok(Some(T::decode(&mut self.body)?))
}
}
pub fn req<T: Decode>(&mut self, field: &'static str) -> Result<T, DecodeError> {
if self.remaining == 0 || peek_code(&self.body) == Some(codes::NULL) {
if self.remaining > 0 {
self.advance_one();
}
return Err(DecodeError::MissingField(field));
}
self.advance_one();
T::decode(&mut self.body)
}
pub fn req_symbols(&mut self, field: &'static str) -> Result<Vec<Symbol>, DecodeError> {
if self.remaining == 0 || peek_code(&self.body) == Some(codes::NULL) {
if self.remaining > 0 {
self.advance_one();
}
return Err(DecodeError::MissingField(field));
}
self.advance_one();
match peek_code(&self.body) {
Some(codes::ARRAY8) | Some(codes::ARRAY32) => {
let (count, body) = read_array_header(&mut self.body)?;
decode_symbol_array(body, count)
}
_ => Ok(vec![Symbol::decode(&mut self.body)?]),
}
}
pub fn symbols(&mut self) -> Result<Vec<Symbol>, DecodeError> {
if self.remaining == 0 {
return Ok(Vec::new());
}
self.advance_one();
match peek_code(&self.body) {
Some(codes::NULL) => {
let _ = read_u8(&mut self.body)?;
Ok(Vec::new())
}
Some(codes::ARRAY8) | Some(codes::ARRAY32) => {
let (count, body) = read_array_header(&mut self.body)?;
decode_symbol_array(body, count)
}
_ => Ok(vec![Symbol::decode(&mut self.body)?]),
}
}
}
#[macro_export]
macro_rules! amqp_composite {
(
$(#[$meta:meta])*
$vis:vis struct $name:ident : $desc:expr => {
$(
$(#[$fmeta:meta])*
$fname:ident : $fty:ty = $kind:ident ( $($karg:tt)* )
),* $(,)?
}
) => {
$(#[$meta])*
#[derive(Debug, Clone, PartialEq)]
#[allow(missing_docs)]
$vis struct $name {
$( $(#[$fmeta])* pub $fname : $fty , )*
}
impl ::core::default::Default for $name {
fn default() -> Self {
Self {
$( $fname : $crate::amqp_composite!(@default $kind ( $($karg)* )) , )*
}
}
}
impl $crate::codec::Encode for $name {
fn encode(&self, buf: &mut ::bytes::BytesMut) {
$crate::codec::encode_described_list(buf, $desc, |fw| {
let _ = &fw;
$( $crate::amqp_composite!(@enc fw, self, $fname, $kind ( $($karg)* )); )*
});
}
}
impl $crate::codec::Decode for $name {
fn decode(buf: &mut ::bytes::Bytes)
-> ::core::result::Result<Self, $crate::codec::DecodeError>
{
#[allow(unused_mut, unused_variables)]
let mut d = $crate::codec::decode_described_list(buf, $desc)?;
Ok(Self {
$( $fname : $crate::amqp_composite!(@dec d, $kind ( $($karg)* )) , )*
})
}
}
};
(@enc $fw:ident, $self:ident, $fname:ident, req ( $($n:tt)* )) => { $fw.field(&$self.$fname); };
(@enc $fw:ident, $self:ident, $fname:ident, opt ( )) => { $fw.field(&$self.$fname); };
(@enc $fw:ident, $self:ident, $fname:ident, default ( $def:expr )) => { $fw.field(&$self.$fname); };
(@enc $fw:ident, $self:ident, $fname:ident, symbols ( )) => { $fw.symbols(&$self.$fname); };
(@enc $fw:ident, $self:ident, $fname:ident, req_symbols ( $($n:tt)* )) => { $fw.symbols_required(&$self.$fname); };
(@default req ( $($n:tt)* )) => { ::core::default::Default::default() };
(@default opt ( )) => { ::core::option::Option::None };
(@default default ( $def:expr )) => { $def };
(@default symbols ( )) => { ::std::vec::Vec::new() };
(@default req_symbols ( $($n:tt)* )) => { ::std::vec::Vec::new() };
(@dec $d:ident, req ( $n:expr )) => { $d.req($n)? };
(@dec $d:ident, opt ( )) => { $d.opt()? };
(@dec $d:ident, default ( $def:expr )) => { $d.opt()?.unwrap_or($def) };
(@dec $d:ident, symbols ( )) => { $d.symbols()? };
(@dec $d:ident, req_symbols ( $n:expr )) => { $d.req_symbols($n)? };
}
fn decode_symbol_array(mut body: Bytes, count: u32) -> Result<Vec<Symbol>, DecodeError> {
if count == 0 {
return Ok(Vec::new());
}
let ctor = read_u8(&mut body)?;
let mut out = Vec::with_capacity((count as usize).min(body.len()));
for _ in 0..count {
let len = match ctor {
codes::SYM8 => read_u8(&mut body)? as usize,
codes::SYM32 => read_u32(&mut body)? as usize,
c => {
return Err(DecodeError::InvalidFormatCode {
code: c,
expected: "symbol array element",
});
}
};
let raw = read_bytes(&mut body, len)?;
let s = String::from_utf8(raw.to_vec())
.map_err(|_| DecodeError::InvalidUtf8 { kind: "symbol" })?;
out.push(Symbol(s));
}
Ok(out)
}