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

//! Flexible serde-based channel for per-call typed communication (Layer 2).

use parking_lot::Mutex;
use serde::{Deserialize, Serialize};
use std::sync::Arc;

use crate::codec::{BincodeCodec, Codec};
use crate::error::{TransportError, TransportResult};
use crate::transport::{FrameTransport, TransportStats};

/// Flexible serde-based channel for per-call typed communication (Layer 2).
#[derive(Debug)]
pub struct SerdeChannel<F: FrameTransport, C: Codec = BincodeCodec> {
    inner: F,
    codec: C,
    stats: Arc<Mutex<TransportStats>>,
}

impl<F: FrameTransport> SerdeChannel<F, BincodeCodec> {
    /// Create a new SerdeChannel with default BincodeCodec.
    pub fn new(transport: F) -> Self {
        Self {
            inner: transport,
            codec: BincodeCodec,
            stats: Arc::new(Mutex::new(TransportStats::default())),
        }
    }
}

impl<F: FrameTransport, C: Codec> SerdeChannel<F, C> {
    /// Create a new SerdeChannel with a specific codec.
    pub fn with_codec(transport: F, codec: C) -> Self {
        Self {
            inner: transport,
            codec,
            stats: Arc::new(Mutex::new(TransportStats::default())),
        }
    }

    /// Send typed data with serialization.
    pub async fn send<T: Serialize>(&self, data: &T) -> TransportResult<()> {
        let encoded = self
            .codec
            .encode(data)
            .map_err(|e| TransportError::Protocol(e.to_string()))?;

        self.inner.send_frame(&encoded).await?;

        let mut stats = self.stats.lock();
        stats.messages_sent += 1;
        stats.bytes_sent += encoded.len() as u64;

        Ok(())
    }

    /// Receive typed data with deserialization.
    pub async fn recv<T: for<'de> Deserialize<'de>>(&self) -> TransportResult<T> {
        let bytes = self.inner.recv_frame().await?;

        let data = self
            .codec
            .decode(&bytes)
            .map_err(|e| TransportError::Protocol(e.to_string()))?;

        let mut stats = self.stats.lock();
        stats.messages_received += 1;
        stats.bytes_received += bytes.len() as u64;

        Ok(data)
    }

    /// Get a reference to the inner transport.
    pub fn inner(&self) -> &F {
        &self.inner
    }

    /// Get transport statistics.
    pub fn stats(&self) -> TransportStats {
        self.stats.lock().clone()
    }
}

// Deprecated alias for backward compatibility
#[deprecated(since = "0.2.0", note = "Use SerdeChannel instead")]
pub type RawTransport<F> = SerdeChannel<F, BincodeCodec>;

#[cfg(test)]
mod tests {
    use super::*;
    use crate::transport::channel::{ChannelConfig, ChannelFrameTransport};

    #[derive(Debug, Clone, PartialEq, serde::Serialize, serde::Deserialize)]
    struct TestRequest {
        id: u64,
        name: String,
    }

    #[tokio::test]
    async fn test_serde_channel_send_recv() {
        let config = ChannelConfig::default();
        let (t1, t2) = ChannelFrameTransport::create_pair("test", config).unwrap();

        let ch1 = SerdeChannel::new(t1);
        let ch2 = SerdeChannel::new(t2);

        let req = TestRequest {
            id: 123,
            name: "test".to_string(),
        };

        ch1.send(&req).await.unwrap();
        let received: TestRequest = ch2.recv().await.unwrap();

        assert_eq!(received, req);
    }

    #[tokio::test]
    async fn test_serde_channel_stats() {
        let config = ChannelConfig::default();
        let (t1, t2) = ChannelFrameTransport::create_pair("test", config).unwrap();

        let ch1 = SerdeChannel::new(t1);
        let ch2 = SerdeChannel::new(t2);

        ch1.send(&42i32).await.unwrap();
        let _: i32 = ch2.recv().await.unwrap();

        let stats1 = ch1.stats();
        assert_eq!(stats1.messages_sent, 1);
        assert!(stats1.bytes_sent > 0);

        let stats2 = ch2.stats();
        assert_eq!(stats2.messages_received, 1);
        assert!(stats2.bytes_received > 0);
    }
}