rsocket_rust 0.7.5

rsocket-rust is an implementation of the RSocket protocol in Rust.
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

use bytes::{Buf, BufMut, Bytes, BytesMut};

use super::{Body, Frame, Version};
use crate::error::RSocketError;
use crate::utils::Writeable;

#[derive(Debug, Eq, PartialEq)]
pub struct Resume {
    version: Version,
    token: Option<Bytes>,
    last_received_server_position: u64,
    first_available_client_position: u64,
}

pub struct ResumeBuilder {
    stream_id: u32,
    flag: u16,
    inner: Resume,
}

impl Resume {
    fn new() -> Resume {
        Resume {
            version: Version::default(),
            token: None,
            last_received_server_position: 0,
            first_available_client_position: 0,
        }
    }

    pub(crate) fn decode(flag: u16, b: &mut BytesMut) -> crate::Result<Resume> {
        if b.len() < 6 {
            return Err(RSocketError::InCompleteFrame.into());
        }
        let major = b.get_u16();
        let minor = b.get_u16();
        let token_size = b.get_u16() as usize;
        let token = if token_size > 0 {
            if b.len() < token_size {
                return Err(RSocketError::InCompleteFrame.into());
            }
            Some(b.split_to(token_size).freeze())
        } else {
            None
        };
        if b.len() < 16 {
            return Err(RSocketError::InCompleteFrame.into());
        }
        let last_received_server_position = b.get_u64();
        let first_available_client_position = b.get_u64();
        Ok(Resume {
            version: Version::new(major, minor),
            token,
            last_received_server_position,
            first_available_client_position,
        })
    }

    pub fn builder(stream_id: u32, flag: u16) -> ResumeBuilder {
        ResumeBuilder::new(stream_id, flag)
    }

    pub fn get_version(&self) -> Version {
        self.version
    }

    pub fn get_token(&self) -> &Option<Bytes> {
        &self.token
    }

    pub fn get_last_received_server_position(&self) -> u64 {
        self.last_received_server_position
    }

    pub fn get_first_available_client_position(&self) -> u64 {
        self.first_available_client_position
    }
}

impl ResumeBuilder {
    fn new(stream_id: u32, flag: u16) -> ResumeBuilder {
        ResumeBuilder {
            stream_id,
            flag,
            inner: Resume::new(),
        }
    }

    pub fn set_token(mut self, token: Bytes) -> Self {
        self.inner.token = Some(token);
        self
    }

    pub fn set_last_received_server_position(mut self, position: u64) -> Self {
        self.inner.last_received_server_position = position;
        self
    }

    pub fn set_first_available_client_position(mut self, position: u64) -> Self {
        self.inner.first_available_client_position = position;
        self
    }

    pub fn build(self) -> Frame {
        Frame {
            stream_id: self.stream_id,
            flag: self.flag,
            body: Body::Resume(self.inner),
        }
    }
}

impl Writeable for Resume {
    fn write_to(&self, bf: &mut BytesMut) {
        self.version.write_to(bf);
        if let Some(b) = self.get_token() {
            bf.put_u16(b.len() as u16);
            bf.extend_from_slice(b);
        }
        bf.put_u64(self.get_last_received_server_position());
        bf.put_u64(self.get_first_available_client_position());
    }

    fn len(&self) -> usize {
        let mut size: usize = 22;
        if let Some(b) = self.get_token() {
            size += b.len();
        }
        size
    }
}