amqpr-codec 0.3.2

Defining AMQP codec
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

use bytes::{BufMut, BytesMut, BigEndian};

use frame::*;


const METHOD_TYPE_BYTE: u8 = 1;
const CONTENT_HEADER_TYPE_BYTE: u8 = 2;
const CONTENT_BODY_TYPE_BYTE: u8 = 3;
const HEARTBEAT_TYPE_BYTE: u8 = 8;


pub fn encode_frame(item: Frame, dst: &mut BytesMut) {
    debug!("Start encode frame : {:?}", item);

    match item.payload {
        FramePayload::Method(payload) => {
            let payload = method::encoder::encode_payload(payload);
            encode_frame_inner(METHOD_TYPE_BYTE, item.header.channel, payload.as_ref(), dst);
        }
        FramePayload::ContentHeader(payload) => {
            let payload = content_header::encode_payload(payload);
            encode_frame_inner(
                CONTENT_HEADER_TYPE_BYTE,
                item.header.channel,
                payload.as_ref(),
                dst,
            );
        }
        FramePayload::ContentBody(payload) => {
            let payload = content_body::encoder::encode_payload(payload);
            encode_frame_inner(
                CONTENT_BODY_TYPE_BYTE,
                item.header.channel,
                payload.as_ref(),
                dst,
            );
        }
        FramePayload::Heartbeat => {
            encode_frame_inner(HEARTBEAT_TYPE_BYTE, item.header.channel, &[], dst);
        }
    };
}


fn encode_frame_inner(type_byte: u8, channel: u16, encoded_payload: &[u8], dst: &mut BytesMut) {
    // put headers
    dst.reserve(7);
    dst.put_u8(type_byte);
    dst.put_u16::<BigEndian>(channel);
    dst.put_u32::<BigEndian>(encoded_payload.len() as u32);
    debug!("byte size of payload is {}", encoded_payload.len());

    // put payload
    dst.extend_from_slice(encoded_payload);

    // put frame-end
    dst.reserve(1);
    dst.put_u8(FRAME_END_OCTET);
}