fibers_rpc 0.2.16

RPC library built on top of fibers crate
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

use bytecodec::bytes::CopyableBytesDecoder;
use bytecodec::marker::Never;
use bytecodec::{self, ByteCount, Decode, Encode, Eos};
use byteorder::{BigEndian, ByteOrder};
use std::cmp;

use message::{MessageHeader, OutgoingMessage};

pub const MIN_PACKET_LEN: usize = PacketHeader::SIZE;
pub const MAX_PACKET_LEN: usize = PacketHeader::SIZE + MAX_PAYLOAD_LEN;
pub const MAX_PAYLOAD_LEN: usize = 0xFFFF;

const FLAG_END_OF_MESSAGE: u8 = 0b0000_0001;
const FLAG_ASYNC: u8 = 0b0000_0010;

#[derive(Debug, Clone)]
pub struct PacketHeader {
    pub message: MessageHeader,
    pub flags: u8,
    pub payload_len: u16,
}
impl PacketHeader {
    pub const SIZE: usize = MessageHeader::SIZE + 1 + 2;

    fn write(&self, buf: &mut [u8]) {
        self.message.write(buf);
        buf[MessageHeader::SIZE] = self.flags;
        BigEndian::write_u16(&mut buf[MessageHeader::SIZE + 1..], self.payload_len);
    }

    fn read(buf: &[u8]) -> Self {
        let mut message = MessageHeader::read(buf);
        let flags = buf[MessageHeader::SIZE];
        message.async = (flags & FLAG_ASYNC) != 0;
        let payload_len = BigEndian::read_u16(&buf[MessageHeader::SIZE + 1..]);
        PacketHeader {
            message,
            flags,
            payload_len,
        }
    }

    pub fn is_end_of_message(&self) -> bool {
        (self.flags & FLAG_END_OF_MESSAGE) != 0
    }

    pub fn is_async(&self) -> bool {
        (self.flags & FLAG_ASYNC) != 0
    }
}

#[derive(Debug, Default)]
pub struct PacketHeaderDecoder {
    bytes: CopyableBytesDecoder<[u8; PacketHeader::SIZE]>,
}
impl Decode for PacketHeaderDecoder {
    type Item = PacketHeader;

    fn decode(&mut self, buf: &[u8], eos: Eos) -> bytecodec::Result<usize> {
        track!(self.bytes.decode(buf, eos))
    }

    fn finish_decoding(&mut self) -> bytecodec::Result<Self::Item> {
        let bytes = track!(self.bytes.finish_decoding())?;
        let header = PacketHeader::read(&bytes[..]);
        Ok(header)
    }

    fn requiring_bytes(&self) -> ByteCount {
        self.bytes.requiring_bytes()
    }

    fn is_idle(&self) -> bool {
        self.bytes.is_idle()
    }
}

#[derive(Debug)]
pub struct PacketizedMessage {
    message: OutgoingMessage,
}
impl PacketizedMessage {
    pub fn new(message: OutgoingMessage) -> Self {
        PacketizedMessage { message }
    }

    pub fn header(&self) -> &MessageHeader {
        &self.message.header
    }
}
impl Encode for PacketizedMessage {
    type Item = Never;

    fn encode(&mut self, buf: &mut [u8], eos: Eos) -> bytecodec::Result<usize> {
        debug_assert!(buf.len() >= PacketHeader::SIZE);

        let limit = cmp::min(buf.len() - PacketHeader::SIZE, MAX_PAYLOAD_LEN);
        let payload_len = track!(
            self.message
                .payload
                .encode(&mut buf[PacketHeader::SIZE..][..limit], eos)
        )?;

        let flags = (self.message.payload.is_idle() as u8 * FLAG_END_OF_MESSAGE)
            | (self.message.header.async as u8 * FLAG_ASYNC);
        let packet_header = PacketHeader {
            message: self.message.header.clone(),
            flags,
            payload_len: payload_len as u16,
        };
        packet_header.write(buf);
        Ok(PacketHeader::SIZE + payload_len)
    }

    fn start_encoding(&mut self, _item: Self::Item) -> bytecodec::Result<()> {
        unreachable!()
    }

    fn is_idle(&self) -> bool {
        self.message.payload.is_idle()
    }

    fn requiring_bytes(&self) -> ByteCount {
        ByteCount::Unknown
    }
}