xCommonLib/protocol/

protocol_v1.rs

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) -> &'a[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);
        }
    }
}