ramqp 0.5.3

A from-scratch, clean-room AMQP 1.0 client.
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
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141

//! Delivery assembly and the lazy, zero-copy [`Delivery`] (WP-4.5, decision D-3).
//!
//! Inbound transfers are assembled into one [`Delivery`] holding the body as
//! [`Bytes`]; typed deserialization is an explicit, cheap opt-in via
//! [`Delivery::decode`] / [`Delivery::message`].

use bytes::{Bytes, BytesMut};

use crate::codec::Decode;
use crate::error::RecvError;
use crate::ids::DeliveryId;
use crate::types::messaging::Message;

/// A received delivery. The body is exposed as raw [`Bytes`] by default; typed
/// access is opt-in (the inversion of `fe2o3-amqp`'s eager-deserialize default).
#[derive(Debug, Clone)]
pub struct Delivery {
    /// The peer-assigned delivery id.
    pub delivery_id: DeliveryId,
    /// The peer-chosen delivery tag.
    pub delivery_tag: Bytes,
    /// Whether the peer pre-settled the delivery.
    pub settled: bool,
    body: Bytes,
}

impl Delivery {
    /// Construct a delivery from assembled parts.
    pub fn new(delivery_id: DeliveryId, delivery_tag: Bytes, settled: bool, body: Bytes) -> Self {
        Delivery {
            delivery_id,
            delivery_tag,
            settled,
            body,
        }
    }

    /// The raw (zero-copy) message bytes.
    pub fn raw(&self) -> &Bytes {
        &self.body
    }

    /// Consume into the raw message bytes.
    pub fn into_raw(self) -> Bytes {
        self.body
    }

    /// Decode the body as a full AMQP [`Message`] (sections + body).
    pub fn message(&self) -> Result<Message, RecvError> {
        let mut buf = self.body.clone();
        Message::decode(&mut buf).map_err(RecvError::from)
    }

    /// Decode the body as a single typed value `T` (e.g. the `amqp-value` body
    /// after stripping sections, when the caller knows the schema).
    pub fn decode<T: Decode>(&self) -> Result<T, RecvError> {
        let mut buf = self.body.clone();
        T::decode(&mut buf).map_err(RecvError::from)
    }
}

/// Accumulates a (possibly multi-frame) delivery as continuation transfers
/// arrive (`more = true` until the final frame).
#[derive(Debug)]
pub struct PartialDelivery {
    delivery_id: DeliveryId,
    delivery_tag: Bytes,
    settled: bool,
    buf: BytesMut,
}

impl PartialDelivery {
    /// Begin a delivery from its first transfer's metadata and payload.
    pub fn new(delivery_id: DeliveryId, delivery_tag: Bytes, settled: bool, first: &[u8]) -> Self {
        let mut buf = BytesMut::with_capacity(first.len());
        buf.extend_from_slice(first);
        PartialDelivery {
            delivery_id,
            delivery_tag,
            settled,
            buf,
        }
    }

    /// The delivery id being assembled.
    pub fn delivery_id(&self) -> DeliveryId {
        self.delivery_id
    }

    /// Bytes accumulated so far (for the message-size cap).
    pub fn len(&self) -> usize {
        self.buf.len()
    }

    /// Whether nothing has been accumulated yet.
    pub fn is_empty(&self) -> bool {
        self.buf.is_empty()
    }

    /// Append a continuation frame's payload.
    pub fn append(&mut self, payload: &[u8]) {
        self.buf.extend_from_slice(payload);
    }

    /// Finish assembly into a [`Delivery`].
    pub fn complete(self) -> Delivery {
        Delivery::new(
            self.delivery_id,
            self.delivery_tag,
            self.settled,
            self.buf.freeze(),
        )
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use crate::codec::to_vec;
    use crate::types::messaging::Message;

    #[test]
    fn lazy_body_and_typed_decode() {
        let msg = Message::text("hello");
        let bytes = Bytes::from(to_vec(&msg));
        let d = Delivery::new(DeliveryId(1), Bytes::from_static(b"tag"), false, bytes.clone());
        assert_eq!(d.raw(), &bytes);
        assert_eq!(d.message().unwrap(), msg);
    }

    #[test]
    fn multi_frame_assembly() {
        let msg = Message::data(Bytes::from_static(b"abcdefgh"));
        let full = to_vec(&msg);
        let (a, b) = full.split_at(full.len() / 2);
        let mut partial = PartialDelivery::new(DeliveryId(2), Bytes::from_static(b"t"), false, a);
        partial.append(b);
        let delivery = partial.complete();
        assert_eq!(delivery.message().unwrap(), msg);
    }
}