xCommonLib/protocol/

protocol_v2.rs

use protobuf::{
    CodedInputStream, CodedOutputStream,
};

use crate::{
    base::status::Status,
    serial::{api_descriptor::make_tag, request_message::RequestMessage},
    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 ProtocolV2Reader<'a> {
    msg_buffer: &'a [u8],
}

impl<'a> ProtocolV2Reader<'a> {
    pub fn new(msg_buffer: &'a [u8]) -> ProtocolV2Reader<'a> {
        ProtocolV2Reader { 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 {
        2
    }
    // 获取连接 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_without_vrs(&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..(self.msg_buffer.len() - 68)]
    }
    //
    pub fn msg_body_with_vrs(&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 fn api(&self) -> Result<String, Status> {
        let msg_body = self.msg_body_without_vrs();

        let mut is = CodedInputStream::from_bytes(msg_body);

        let tag = is.read_raw_tag_or_eof();

        if let Err(err) = tag {
            return Err(Status::error(err.to_string()));
        }

        let tag = tag.unwrap();

        if tag.is_none() {
            return Err(Status::error("读取 Api 类型标签出错！".into()));
        }

        let tag = tag.unwrap();
        let api_tag = make_tag(1, 2);
        if tag != api_tag {
            return Err(Status::error("Api 类型错误！".into()));
        }

        let api = is.read_string();

        match api {
            Ok(api) => Ok(api),
            Err(err) => Err(Status::error(err.to_string())),
        }
    }
}

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

impl ProtocolV2Writer {
    pub fn new(
        msg_size: u32,
        req_id: i64,
        sender: &ServiceKey,
        receiver: &ServiceKey,
    ) -> ProtocolV2Writer {
        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 = ProtocolV2Writer {
            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] {
        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 = (2 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);
        }
    }
}