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
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
use protobuf::CodedOutputStream;

use crate::{serial::request_message::RequestMessage};
use crate::service::sys_service_api::ServiceKey;
use super::{MsgType, XID};

const DATA_LEN_POS: u16 = 0;
const VERSION_POS: u16 = 4;
const MSG_TYPE_POS: u16 = VERSION_POS + 2;
const CONN_ID_POS: u16 = MSG_TYPE_POS + 4;
const REQ_ID_POS: u16 = CONN_ID_POS + 8;
const TIME_OUT_POS: u16 = REQ_ID_POS + 8;
const FIX_LEN_POS: u16 = TIME_OUT_POS + 4;
// len + type

pub struct ProtocolV1Reader<'a> {
    msg_buffer: &'a [u8],
}

impl<'a> ProtocolV1Reader<'a> {
    pub fn new(msg_buffer: &'a [u8]) -> ProtocolV1Reader<'a> {
        ProtocolV1Reader { msg_buffer }
    }

    pub fn msg_type(&self) -> Option<MsgType> {
        unsafe {
            let ptr = self.msg_buffer.as_ptr().offset(MSG_TYPE_POS as isize);

            let bytes = *(ptr as *const [u8; 4]);

            let value = i32::from_le_bytes(bytes);

            MsgType::from(value)
        }
    }
    //
    pub fn version() -> u16 {
        1
    }
    // 获取连接 id  = 4 字节 ip + 2 字节 端口号
    pub fn conn_id(&self) -> i64 {
        unsafe {
            let ptr = self.msg_buffer.as_ptr().offset(CONN_ID_POS as isize);

            let bytes = *(ptr as *const [u8; 8]);

            i64::from_le_bytes(bytes)
        }
    }

    /**
     * 获取当前请求的 id
     */
    pub fn req_id(&self) -> i64 {
        unsafe {
            let ptr = self.msg_buffer.as_ptr().offset(REQ_ID_POS as isize);

            let bytes = *(ptr as *const [u8; 8]);

            i64::from_le_bytes(bytes)
        }
    }

    /**
     *
     */
    pub fn xid(&self) -> XID {
        XID {
            conn_id: self.conn_id(),
            request_id: self.req_id(),
        }
    }
    /**
     * 超时
     */
    pub fn timeout(&self) -> u32 {
        unsafe {
            let ptr = self.msg_buffer.as_ptr().offset(TIME_OUT_POS as isize);

            let bytes = *(ptr as *const [u8; 4]);

            u32::from_le_bytes(bytes)
        }
    }
    /**
     * 获取消息长度
     */
    pub fn msg_size(&self) -> u32 {
        self.msg_buffer.len() as u32
    }

    pub fn msg_body(&self) -> &[u8] {
        let sender_key_slice = &self.msg_buffer[FIX_LEN_POS as usize..];

        let skip_sender_bytes = ServiceKey::skip_from_bytes(sender_key_slice);

        let skip_bytes = FIX_LEN_POS as usize + skip_sender_bytes as usize;

        let receiver_key_slice = &self.msg_buffer[skip_bytes..];

        let skip_receiver_bytes = ServiceKey::skip_from_bytes(receiver_key_slice);

        &self.msg_buffer[(skip_bytes + skip_receiver_bytes as usize) as usize..]
    }

    pub fn sender(&self) -> ServiceKey {
        let service_key_slice = &self.msg_buffer[FIX_LEN_POS as usize..];

        let mut sender = ServiceKey::default();

        sender
            .parse_from_bytes_return_num(service_key_slice)
            .unwrap();

        sender
    }

    pub fn receiver(&self) -> ServiceKey {
        let sender_key_slice = &self.msg_buffer[FIX_LEN_POS as usize..];

        let skip_sender_bytes = ServiceKey::skip_from_bytes(sender_key_slice);

        let mut receiver = ServiceKey::default();

        let skip_bytes = FIX_LEN_POS as usize + skip_sender_bytes as usize;

        let service_key_slice = &self.msg_buffer[skip_bytes..];

        receiver
            .parse_from_bytes_return_num(service_key_slice)
            .unwrap();

        receiver
    }
}

pub struct ProtocolV1Writer {
    pub msg_buffer: Vec<u8>,
    pub msg_body_size: u32,
}

impl ProtocolV1Writer {
    pub fn new(
        msg_size: u32,
        req_id: i64,
        sender: &ServiceKey,
        receiver: &ServiceKey,
    ) -> ProtocolV1Writer {
        let sender_size = sender.compute_size_with_tag_and_len() as u32;
        
        let receiver_size = receiver.compute_size_with_tag_and_len() as u32;

        let msg_len = (FIX_LEN_POS as u32 + msg_size + sender_size + receiver_size as u32) as usize;

        let mut msg_buffer: Vec<u8> = Vec::with_capacity(msg_len);
        unsafe {
            msg_buffer.set_len(msg_len);
        }

        let mut writer = ProtocolV1Writer {
            msg_buffer,
            msg_body_size: msg_size,
        };
        //  写入 xid
        writer.write_conn_id(0);
        writer.write_req_id(req_id);

        writer.write_version();
        writer.write_data_len(msg_len as u32);
        writer.write_timeout(u32::MAX);
        //

        writer.write_service_key(sender, receiver);

        writer
    }

    fn write_service_key(&mut self, sender: &ServiceKey, receiver: &ServiceKey) {
        //
        unsafe {
            let service_key_slice = self.msg_buffer.as_mut_ptr().offset(FIX_LEN_POS as isize);

            let service_key_slice = std::slice::from_raw_parts_mut(
                service_key_slice,
                self.msg_buffer.len() - FIX_LEN_POS as usize,
            );

            let mut os = CodedOutputStream::bytes(service_key_slice);

            sender.serial_with_tag_and_len(&mut os);

            receiver.serial_with_tag_and_len(&mut os);

        }
    }

    pub fn write_req_id(&mut self, req_id: i64) {
        unsafe {
            let bytes = req_id.to_le_bytes();

            let ptr = self.msg_buffer.as_mut_ptr().offset(REQ_ID_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 8);
        }
    }
    /**
     *
     */
    pub fn write_conn_id(&mut self, conn_id: i64) {
        unsafe {
            let bytes = conn_id.to_le_bytes();

            let ptr = self.msg_buffer.as_mut_ptr().offset(CONN_ID_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 8);
        }
    }

    pub fn write_conn_id_to_message(message: &mut Vec<u8>, conn_id: i64) {
        unsafe {
            let bytes = conn_id.to_le_bytes();

            let ptr = message.as_mut_ptr().offset(CONN_ID_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 8);
        }
    }

    pub fn write_msg_type(&mut self, msg_type: MsgType) {
        unsafe {
            let bytes = (msg_type as i32).to_le_bytes();

            let ptr = self.msg_buffer.as_mut_ptr().offset(MSG_TYPE_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 4);
        }
    }

    pub fn msg_body_buffer(&mut self) -> &mut [u8] {
        // msg_body_size

        let msg_body_pos = self.msg_buffer.len() - self.msg_body_size as usize;

        &mut self.msg_buffer[msg_body_pos as usize..]
    }

    pub fn write_msg_body(&mut self, message: &Vec<u8>) {
        unsafe {
            let msg_body_pos = self.msg_buffer.len() - self.msg_body_size as usize;

            let dst_ptr = self.msg_buffer.as_mut_ptr().offset(msg_body_pos as isize);

            std::ptr::copy_nonoverlapping(message.as_ptr(), dst_ptr, message.len());
        }
    }

    pub fn write_msg_body_by_arr(&mut self, message: &[u8]) {
      unsafe {
          let msg_body_pos = self.msg_buffer.len() - self.msg_body_size as usize;

          let dst_ptr = self.msg_buffer.as_mut_ptr().offset(msg_body_pos as isize);

          std::ptr::copy_nonoverlapping(message.as_ptr(), dst_ptr, message.len());
      }
  }

    /**
     *
     */
    pub fn write_timeout(&mut self, timeout: u32) {
        unsafe {
            let bytes = timeout.to_le_bytes();

            let ptr = self.msg_buffer.as_mut_ptr().offset(TIME_OUT_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 4);
        }
    }

    /**
     *
     */
    fn write_version(&mut self) {
        unsafe {
            let bytes = (1 as u16).to_le_bytes();

            let ptr = self.msg_buffer.as_mut_ptr().offset(VERSION_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 2);
        }
    }
    /**
     * 写入长度
     */
    fn write_data_len(&mut self, data_len: u32) {
        unsafe {
            let bytes = data_len.to_le_bytes();

            let ptr = self.msg_buffer.as_mut_ptr().offset(DATA_LEN_POS as isize);

            std::ptr::copy_nonoverlapping(bytes.as_ptr(), ptr, 4);
        }
    }
}