amq_protocol/frame/
generation.rs

1use crate::{
2    frame::{AMQPFrame, ProtocolVersion},
3    protocol::{basic::gen_properties, *},
4    types::{generation::*, *},
5};
6use cookie_factory::{combinator::slice, sequence::tuple};
7use std::io::Write;
8
9/// Serialize a frame in the given buffer
10pub fn gen_frame<'a, W: Write + BackToTheBuffer + 'a>(
11    frame: &'a AMQPFrame,
12) -> impl SerializeFn<W> + 'a {
13    move |x| match frame {
14        AMQPFrame::ProtocolHeader(version) => gen_protocol_header(*version)(x),
15        AMQPFrame::Heartbeat => gen_heartbeat_frame(0)(x),
16        AMQPFrame::InvalidHeartbeat(channel_id) => gen_heartbeat_frame(*channel_id)(x),
17        AMQPFrame::Method(channel_id, method) => gen_method_frame(*channel_id, method)(x),
18        AMQPFrame::Header(channel_id, header) => gen_content_header_frame(
19            *channel_id,
20            header.class_id,
21            header.body_size,
22            &header.properties,
23        )(x),
24        AMQPFrame::Body(channel_id, data) => gen_content_body_frame(*channel_id, data)(x),
25    }
26}
27
28fn gen_protocol_header<W: Write>(version: ProtocolVersion) -> impl SerializeFn<W> {
29    tuple((
30        slice(metadata::NAME.as_bytes()),
31        gen_short_short_uint(0),
32        gen_protocol_version(version),
33    ))
34}
35
36fn gen_protocol_version<W: Write>(version: ProtocolVersion) -> impl SerializeFn<W> {
37    tuple((
38        gen_short_short_uint(version.major),
39        gen_short_short_uint(version.minor),
40        gen_short_short_uint(version.revision),
41    ))
42}
43
44fn gen_heartbeat_frame<W: Write>(channel_id: ChannelId) -> impl SerializeFn<W> {
45    tuple((
46        gen_short_short_uint(constants::FRAME_HEARTBEAT),
47        gen_id(channel_id),
48        gen_long_uint(0),
49        gen_short_short_uint(constants::FRAME_END),
50    ))
51}
52
53fn gen_method_frame<'a, W: Write + BackToTheBuffer + 'a>(
54    channel_id: ChannelId,
55    class: &'a AMQPClass,
56) -> impl SerializeFn<W> + 'a {
57    tuple((
58        gen_short_short_uint(constants::FRAME_METHOD),
59        gen_id(channel_id),
60        gen_with_len(gen_class(class)),
61        gen_short_short_uint(constants::FRAME_END),
62    ))
63}
64
65fn gen_content_header_frame<'a, W: Write + BackToTheBuffer + 'a>(
66    channel_id: ChannelId,
67    class_id: Identifier,
68    length: PayloadSize,
69    properties: &'a basic::AMQPProperties,
70) -> impl SerializeFn<W> + 'a {
71    tuple((
72        gen_short_short_uint(constants::FRAME_HEADER),
73        gen_id(channel_id),
74        gen_with_len(tuple((
75            gen_id(class_id),
76            gen_short_uint(0 /* weight */),
77            gen_long_long_uint(length),
78            gen_properties(properties),
79        ))),
80        gen_short_short_uint(constants::FRAME_END),
81    ))
82}
83
84fn gen_content_body_frame<'a, W: Write + 'a>(
85    channel_id: ChannelId,
86    content: &'a [u8],
87) -> impl SerializeFn<W> + 'a {
88    tuple((
89        gen_short_short_uint(constants::FRAME_BODY),
90        gen_id(channel_id),
91        gen_long_uint(content.len() as ChunkSize),
92        slice(content),
93        gen_short_short_uint(constants::FRAME_END),
94    ))
95}
96
97#[cfg(test)]
98mod test {
99    use super::*;
100
101    #[test]
102    fn generate_header_frame() {
103        use crate::{
104            frame::{AMQPContentHeader, WriteContext},
105            protocol::BasicProperties,
106        };
107
108        let channel_id = 1;
109        let hdr = AMQPContentHeader {
110            class_id: 60,
111            body_size: 5,
112            properties: BasicProperties::default(),
113        };
114        let header = AMQPFrame::Header(channel_id, hdr);
115
116        let buf = Vec::<u8>::new();
117        let val = gen_frame::<Vec<u8>>(&header);
118
119        let ctx = WriteContext::from(buf);
120        let (frame, _size) = val(ctx).unwrap().into_inner();
121        println!("header: {:?}", header);
122        println!("frame: {:?}", frame);
123
124        let expected = [
125            2, // frame type
126            0, 1, // channel ID
127            0, 0, 0, 14, // payload size (of header frame)
128            0, 60, // <-- method class id, not serialized, but should
129            0, 0, // weight, basically unused
130            0, 0, 0, 0, 0, 0, 0, 5, // body_size
131            0, 0,   // property flags
132            206, // 0xCE frame end marker
133        ];
134
135        assert_eq!(frame, expected);
136    }
137}
amq_protocol/frame/generation.rs

amq_protocol/frame/
generation.rs
