tonic-server-dispatch 0.5.0

A request-dispatch level for tonic-Server.
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

/// internal macro
///
// define tonic Server: [<$service DispatchServer>]
//
// used by sync and async modes both
#[macro_export]
macro_rules! _define_dispatch_server {
    (
        $service:ty,
        $hash_by:ident : $hash_type:ty,

        $mpsc_sender_type:ty,

        [ $(
            $shard_mutable_method:ident ($shard_mutable_request:ty) -> $shard_mutable_reply:ty,
        )* ],

        [ $(
            $shard_readonly_method:ident ($shard_readonly_request:ty) -> $shard_readonly_reply:ty,
        )* ],

        [ $(
            $item_mutable_method:ident ($item_mutable_request:ty) -> $item_mutable_reply:ty,
        )* ],

        [ $(
            $item_readonly_method:ident ($item_readonly_request:ty) -> $item_readonly_reply:ty,
        )* ]
    ) => {

        paste::paste! {

        // Context for the tonic server, used to dispatch requests.
        pub struct [<$service DispatchServer>] {
            txs: Vec<$mpsc_sender_type<DispatchRequest>>,
        }

        impl [<$service DispatchServer>] {
            // create with txs
            fn with_txs(txs: Vec<$mpsc_sender_type<DispatchRequest>>) -> Self {
                Self { txs }
            }
        }

        // The tonic server implementation.
        //
        // Dispatch the request to backend, and wait for the reply.
        #[tonic::async_trait]
        impl $service for [<$service DispatchServer>] {
             $(
                async fn $shard_mutable_method(
                    &self,
                    request: tonic::Request<$shard_mutable_request>,
                ) -> Result<tonic::Response<$shard_mutable_reply>, tonic::Status> {
                    tonic_server_dispatch::_service_method_body!($shard_mutable_method, self, request, $hash_by)
                }
             )*
             $(
                async fn $shard_readonly_method(
                    &self,
                    request: tonic::Request<$shard_readonly_request>,
                ) -> Result<tonic::Response<$shard_readonly_reply>, tonic::Status> {
                    tonic_server_dispatch::_service_method_body!($shard_readonly_method, self, request, $hash_by)
                }
             )*
             $(
                async fn $item_mutable_method(
                    &self,
                    request: tonic::Request<$item_mutable_request>,
                ) -> Result<tonic::Response<$item_mutable_reply>, tonic::Status> {
                    tonic_server_dispatch::_service_method_body!($item_mutable_method, self, request, $hash_by)
                }
            )*
            $(
                async fn $item_readonly_method(
                    &self,
                    request: tonic::Request<$item_readonly_request>,
                ) -> Result<tonic::Response<$item_readonly_reply>, tonic::Status> {
                    tonic_server_dispatch::_service_method_body!($item_readonly_method, self, request, $hash_by)
                }
            )*
        }

        }
    }
}

/// internal macro
///
// service method body, for shard_method, mutable_method and readonly_method all.
//
// dispatch requests to backends and wait for repsponse.
#[macro_export]
macro_rules! _service_method_body {
    ($method:ident, $self:ident, $request:expr, $hash_by:ident) => {
        paste::paste! {
            {
                fn calc_hash(item: &impl std::hash::Hash) -> u64 {
                    use std::hash::Hasher;
                    let mut hasher = std::collections::hash_map::DefaultHasher::new();
                    item.hash(&mut hasher);
                    hasher.finish()
                }

                let request = $request.into_inner();

                let shard = calc_hash(&request.$hash_by) as usize % $self.txs.len();

                let (resp_tx, resp_rx) = tokio::sync::oneshot::channel();

                let biz_req = DispatchRequest::[<$method:camel>](request, resp_tx);

                match $self.txs[shard].try_send(biz_req) {
                    Ok(()) => resp_rx.await.unwrap().map(tonic::Response::new),
                    Err(_) => Err(tonic::Status::unavailable(String::new())),
                }
            }
        }
    };
}