occams-rpc 0.3.0

razor-rpc is a modular, pluggable RPC for high throughput scenario, supports various runtimes, with a low-level streaming interface, and high-level remote API call interface.
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

//! The dispatch layer for API server

use super::service::ServiceStatic;
use super::task::{APIServerReq, APIServerResp};
use captains_log::*;
use razor_stream::{
    Codec,
    proto::RpcAction,
    server::{RpcSvrReq, dispatch::Dispatch, task::RespNoti},
};
use std::marker::PhantomData;
use std::sync::Arc;

/// Interface for all dispatch for API Server
pub trait APIDispatchTrait: Send + Sync + 'static + Clone {
    type Codec: Codec;

    fn dispatch(
        &self, req: APIServerReq<Self::Codec>,
    ) -> impl Future<Output = Result<(), ()>> + Send;
}

/// A container for everything impl APIDispatchTrait
///
/// Because rust orphan rule forbid blanet impl non-local trait
#[derive(Clone)]
pub struct APIDispatch<D: APIDispatchTrait>(D);

impl<D: APIDispatchTrait> Dispatch for APIDispatch<D> {
    type RespTask = APIServerResp;

    type Codec = D::Codec;

    #[inline]
    async fn dispatch_req<'a>(
        &'a self, codec: &Arc<Self::Codec>, req: RpcSvrReq<'a>, noti: RespNoti<Self::RespTask>,
    ) -> Result<(), ()> {
        if let RpcAction::Str(action) = req.action {
            if let Some((service, method)) = action.split_once('.') {
                return self
                    .0
                    .dispatch(APIServerReq::<Self::Codec> {
                        seq: req.seq,
                        service: service.to_string(),
                        method: method.to_string(),
                        req: Some(req.msg.to_vec()),
                        codec: codec.clone(),
                        noti,
                    })
                    .await;
            }
        }
        warn!("{:?} invalid action", req);
        return Err(());
    }
}

pub type DispatchInline<C, S> = APIDispatch<Inline<C, S>>;

/// APIDispatch for inline process in connection coroutine, only for demo
///
/// It will block the next request if your async method blocks.
pub struct Inline<C: Codec, S: ServiceStatic<C> + Clone> {
    service: S,
    _phan: PhantomData<fn(&C)>,
}

impl<C: Codec, S: ServiceStatic<C> + Clone> Inline<C, S> {
    #[inline]
    pub fn new(s: S) -> DispatchInline<C, S> {
        APIDispatch(Inline { service: s, _phan: Default::default() })
    }
}

impl<C: Codec, S: ServiceStatic<C> + Clone> Clone for Inline<C, S> {
    #[inline]
    fn clone(&self) -> Self {
        Self { service: self.service.clone(), _phan: Default::default() }
    }
}

impl<C: Codec, S: ServiceStatic<C> + Clone> APIDispatchTrait for Inline<C, S> {
    type Codec = C;

    #[inline]
    async fn dispatch(&self, req: APIServerReq<C>) -> Result<(), ()> {
        self.service.serve(req).await;
        return Ok(());
    }
}