xy-rpc 0.1.0

An RPC framework for Rust
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

use crate::formats::SerdeFormat;
use crate::{ChannelBuilder, RpcError, RpcFrameHead, RpcMsgHandler, RpcMsgHandlerWrapper, RpcServiceSchema, XyRpcChannel, new_transport_sink, new_transport_stream, ServiceFactory};
use std::future::Future;
use tokio::io::{AsyncRead, AsyncReadExt, AsyncWrite, AsyncWriteExt};
use tokio_util::compat::{TokioAsyncReadCompatExt, TokioAsyncWriteCompatExt};

pub trait ChannelBuilderTokioExt<SF, CS, MH, MSG> {
    fn build_from_tokio(
        self,
        io: impl AsyncRead + AsyncWrite + Unpin + Send + 'static,
    ) -> (
        XyRpcChannel<SF, CS>,
        impl Future<Output = Result<(), RpcError>> + Send + 'static,
    )
    where
        XyRpcChannel<SF, CS>: Clone,
        CS: RpcServiceSchema,
        MH: ServiceFactory<SF, CS>;
    fn build_from_tokio_read_write<S: RpcServiceSchema>(
        self,
        io: (
            impl AsyncRead + Unpin + Send + 'static,
            impl AsyncWrite + Unpin + Send + 'static,
        ),
    ) -> (
        XyRpcChannel<SF, CS>,
        impl Future<Output = Result<(), RpcError>> + Send + 'static,
    )
    where
        XyRpcChannel<SF, CS>: Clone,
        CS: RpcServiceSchema,
        MH: ServiceFactory<SF, CS>;
}
impl<SF, CS, MH, MSG> ChannelBuilderTokioExt<SF, CS, MH, MSG> for ChannelBuilder<SF, CS, MH, MSG>
where
    SF: SerdeFormat,
{
    fn build_from_tokio(
        self,
        io: impl AsyncRead + AsyncWrite + Unpin + Send + 'static,
    ) -> (
        XyRpcChannel<SF, CS>,
        impl Future<Output = Result<(), RpcError>> + Send + 'static,
    )
    where
        XyRpcChannel<SF, CS>: Clone,
        CS: RpcServiceSchema,
        MH: ServiceFactory<SF, CS>,
    {
        self.build_from_tokio_read_write::<MH::S>(tokio::io::split(io))
    }
    fn build_from_tokio_read_write<S: RpcServiceSchema>(
        self,
        (read, write): (
            impl AsyncRead + Unpin + Send + 'static,
            impl AsyncWrite + Unpin + Send + 'static,
        ),
    ) -> (
        XyRpcChannel<SF, CS>,
        impl Future<Output = Result<(), RpcError>> + Send + 'static,
    )
    where
        XyRpcChannel<SF, CS>: Clone,
        CS: RpcServiceSchema,
        MH: ServiceFactory<SF, CS>,
    {
        let stream = new_transport_stream(read.compat());
        let sink = new_transport_sink(write.compat_write());
        self.build_from_transport(sink, stream)
    }
}

pub async fn write_frame(
    mut write: impl AsyncWrite + Unpin,
    frame: &RpcFrameHead,
) -> Result<(), RpcError> {
    let bits: u64 = frame.into();
    write.write_u64(bits).await?;
    Ok(())
}

pub async fn read_frame(mut read: impl AsyncRead + Unpin) -> Result<RpcFrameHead, RpcError> {
    let bits = read.read_u64().await?;
    bits.try_into()
}

#[cfg(feature = "duplex")]
pub async fn serve_duplex<
    SF: SerdeFormat,
    T1: 'static,
    T2: 'static,
    O1: Send + 'static,
    O2: Send + 'static,
    CS1: RpcServiceSchema + Send + 'static,
    CS2: RpcServiceSchema + Send + 'static,
    F1: Future<Output = Result<O1, RpcError>> + Send + 'static,
    F2: Future<Output = Result<O2, RpcError>> + Send + 'static,
>(
    serde_format: SF,
    (serve1, f1): (
        impl FnOnce(XyRpcChannel<SF, CS2>) -> T1,
        impl FnMut(XyRpcChannel<SF, CS2>) -> F1 + Send + 'static,
    ),
    (serve2, f2): (
        impl FnOnce(XyRpcChannel<SF, CS1>) -> T2,
        impl FnMut(XyRpcChannel<SF, CS1>) -> F2 + Send + 'static,
    ),
) -> Result<(O1, O2), RpcError>
where
    RpcMsgHandlerWrapper<T1>: RpcMsgHandler<CS1>,
    RpcMsgHandlerWrapper<T2>: RpcMsgHandler<CS2>,
{
    let (duplex1, duplex2) = tokio::io::duplex(1024 * 8);
    let (read1, write1) = tokio::io::split(duplex1);
    let (read2, write2) = tokio::io::split(duplex2);
    crate::duplex::serve_duplex_fromserve_duplex_from(
        (
            (read1.compat(), write1.compat_write()),
            (read2.compat(), write2.compat_write()),
        ),
        serde_format,
        (serve1, f1),
        (serve2, f2),
    )
    .await
}