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

use std::convert::TryInto;
use std::sync::Arc;

//Message used in event bus in standalone instance and cluster
#[derive(Clone, Default, Debug)]
pub struct Message {
    //Destination sub address
    pub(crate) address: Option<String>,
    //Replay sub address
    pub(crate) replay: Option<String>,
    //Binary body content
    pub(crate) body: Arc<Vec<u8>>,
    //Protocol version
    pub(crate) protocol_version: i32,
    //System codec id
    pub(crate) system_codec_id: i32,
    //Port to replay message
    pub(crate) port: i32,
    //Host to replay message
    pub(crate) host: String,
    //Headers
    pub(crate) headers: i32,
    //Message send as publish to all nodes in sub
    pub(crate) publish: bool,
}

impl Message {
    #[inline]
    pub fn body(&self) -> Arc<Vec<u8>> {
        return self.body.clone();
    }

    //Reply message to event bus
    #[inline]
    pub fn reply(&mut self, mut data: Vec<u8>) {
        unsafe {
            let mut clone_body = self.body.clone();
            let inner_body = Arc::get_mut_unchecked(&mut clone_body);
            inner_body.clear();
            inner_body.append(&mut data);
            self.address = self.replay.clone();
            self.replay = None;
        }
    }

    pub fn generate() -> Message {
        Message {
            address: Some("test.01".to_string()),
            replay: Some(format!(
                "__vertx.reply.{}",
                uuid::Uuid::new_v4().to_string()
            )),
            body: Arc::new(uuid::Uuid::new_v4().to_string().into_bytes()),
            port: 44532 as i32,
            host: "localhost".to_string(),
            ..Default::default()
        }
    }
}

//Implementation of deserialize byte array to message
impl From<Vec<u8>> for Message {
    #[inline]
    fn from(msg: Vec<u8>) -> Self {
        let mut idx = 3; //Ignore first 3 bytes
        let len_addr = i32::from_be_bytes(msg[idx..idx + 4].try_into().unwrap()) as usize;
        idx += 4;
        let address = String::from_utf8(msg[idx..idx + len_addr].to_vec()).unwrap();
        idx += len_addr;
        let len_replay = i32::from_be_bytes(msg[idx..idx + 4].try_into().unwrap()) as usize;
        idx += 4;
        let mut replay = None;
        if len_replay > 0 {
            let replay_str = String::from_utf8(msg[idx..idx + len_addr].to_vec()).unwrap();
            idx += len_replay;
            replay = Some(replay_str);
        }
        let port = i32::from_be_bytes(msg[idx..idx + 4].try_into().unwrap());
        idx += 4;
        let len_host = i32::from_be_bytes(msg[idx..idx + 4].try_into().unwrap()) as usize;
        idx += 4;
        let host = String::from_utf8(msg[idx..idx + len_host].to_vec()).unwrap();
        idx += len_host;
        let headers = i32::from_be_bytes(msg[idx..idx + 4].try_into().unwrap());
        idx += 4;
        let len_body = i32::from_be_bytes(msg[idx..idx + 4].try_into().unwrap()) as usize;
        idx += 4;
        let body = msg[idx..idx + len_body].to_vec();
        Message {
            address: Some(address.to_string()),
            replay,
            port,
            host,
            headers,
            body: Arc::new(body),
            ..Default::default()
        }
    }
}

impl Message {
    //Serialize message to byte array
    #[inline]
    pub fn to_vec(&self) -> Result<Vec<u8>, &str> {
        let mut data = vec![];
        data.push(1);
        data.push(12);
        data.push(0);
        let address = self.address.clone().expect("Replay message not found!");
        data.extend_from_slice(&(address.len() as i32).to_be_bytes());
        data.extend_from_slice(address.as_bytes());
        match self.replay.clone() {
            Some(addr) => {
                data.extend_from_slice(&(addr.len() as i32).to_be_bytes());
                data.extend_from_slice(addr.as_bytes());
            }
            None => {
                data.extend_from_slice(&(0 as i32).to_be_bytes());
            }
        }
        data.extend_from_slice(&self.port.to_be_bytes());
        data.extend_from_slice(&(self.host.len() as i32).to_be_bytes());
        data.extend_from_slice(self.host.as_bytes());
        data.extend_from_slice(&(4 as i32).to_be_bytes());
        data.extend_from_slice(&(self.body.len() as i32).to_be_bytes());
        data.extend_from_slice(self.body.as_slice());
        let len = ((data.len()) as i32).to_be_bytes();
        for idx in 0..4 {
            data.insert(idx, len[idx]);
        }
        return Ok(data);
    }
}