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;
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
}
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)
}
}
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,
};
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] {
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);
}
}
}