xrpc-rs 0.3.0

async RPC library optimized for shared memory communication
Documentation 
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

use serde::{Deserialize, Serialize};
use std::sync::atomic::{AtomicU64, Ordering};

use crate::error::{Result, RpcError};

pub const MAGIC: [u8; 4] = [0x58, 0x52, 0x50, 0x43]; // XRPC
pub const VERSION: u8 = 1;
pub const MAX_MESSAGE_SIZE: usize = 16 * 1024 * 1024;
pub const MIN_HEADER_SIZE: usize = 4 + 1 + 1 + 4 + 8 + 1 + 2 + 4 + 4;

static NEXT_MESSAGE_ID: AtomicU64 = AtomicU64::new(1);

#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash, Serialize, Deserialize)]
pub struct MessageId(pub u64);

impl MessageId {
    pub fn new() -> Self {
        MessageId(NEXT_MESSAGE_ID.fetch_add(1, Ordering::Relaxed))
    }

    pub fn from_raw(id: u64) -> Self {
        MessageId(id)
    }

    pub fn raw(&self) -> u64 {
        self.0
    }
}

impl Default for MessageId {
    fn default() -> Self {
        Self::new()
    }
}

impl std::fmt::Display for MessageId {
    fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
        write!(f, "{}", self.0)
    }
}

#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[repr(u8)]
pub enum MessageType {
    Call = 0,
    Reply = 1,
    Notification = 2,
    Error = 3,
    StreamChunk = 4,
    StreamEnd = 5,
}

impl MessageType {
    pub fn from_u8(value: u8) -> Result<Self> {
        match value {
            0 => Ok(MessageType::Call),
            1 => Ok(MessageType::Reply),
            2 => Ok(MessageType::Notification),
            3 => Ok(MessageType::Error),
            4 => Ok(MessageType::StreamChunk),
            5 => Ok(MessageType::StreamEnd),
            _ => Err(RpcError::InvalidMessage(format!(
                "Unknown message type: {}",
                value
            ))),
        }
    }

    pub fn to_u8(self) -> u8 {
        self as u8
    }
}

/// Compression algorithm used for message payload
#[derive(Debug, Clone, Copy, PartialEq, Eq, Serialize, Deserialize)]
#[repr(u8)]
#[derive(Default)]
pub enum CompressionType {
    #[default]
    None = 0,
    Lz4 = 1,
    Zstd = 2,
}

impl CompressionType {
    pub fn from_u8(value: u8) -> Result<Self> {
        match value {
            0 => Ok(CompressionType::None),
            1 => Ok(CompressionType::Lz4),
            2 => Ok(CompressionType::Zstd),
            _ => Err(RpcError::InvalidMessage(format!(
                "Unknown compression type: {}",
                value
            ))),
        }
    }

    pub fn to_u8(self) -> u8 {
        self as u8
    }
}

/// Message flags packed into a single byte
#[derive(Debug, Clone, Copy, Default)]
pub struct MessageFlags {
    /// Whether the message payload is compressed
    pub compressed: bool,
    /// Whether this is a streaming message
    pub streaming: bool,
    /// Whether this is part of a batch
    pub batch: bool,
}

impl MessageFlags {
    pub fn from_u8(value: u8) -> Self {
        Self {
            compressed: (value & 0x01) != 0,
            streaming: (value & 0x02) != 0,
            batch: (value & 0x04) != 0,
        }
    }

    pub fn to_u8(self) -> u8 {
        let mut flags = 0u8;
        if self.compressed {
            flags |= 0x01;
        }
        if self.streaming {
            flags |= 0x02;
        }
        if self.batch {
            flags |= 0x04;
        }
        flags
    }
}