pajamax 0.3.1

Fast gRPC server framework in synchronous mode.
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

use std::cell::RefCell;
use std::net::TcpStream;
use std::sync::{mpsc, Arc, Mutex};

use crate::config::Config;
use crate::connection::local_build_response;
use crate::error::Error;
use crate::response_end::ResponseEnd;
use crate::status::{Code, Status};
use crate::ReplyEncode;
use crate::Response;

/// Send end of request channel for dispatch mode.
pub type RequestTx<Req> = mpsc::SyncSender<DispatchRequest<Req>>;

/// Receive end of request channel for dispatch mode.
pub type RequestRx<Req> = mpsc::Receiver<DispatchRequest<Req>>;

/// Send end of response channel for dispatch mode.
type ResponseTx = mpsc::SyncSender<DispatchResponse>;

/// Receive end of response channel for dispatch mode.
type ResponseRx = mpsc::Receiver<DispatchResponse>;

/// Dispatched request in dispatch mode.
pub struct DispatchRequest<Req> {
    pub stream_id: u32,
    pub req_data_len: usize,
    pub request: Req,
    pub resp_tx: ResponseTx,
}

/// Dispatched response in dispatch mode.
pub struct DispatchResponse {
    pub stream_id: u32,
    pub req_data_len: usize,

    // We use dynamic-dispatch `dyn` here to accept different
    // response from multiple services in one channel.
    pub response: Response<Box<dyn ReplyEncode>>,
}

thread_local! {
    static RESP_TX: RefCell<ResponseTx> = panic!();
}

// create a backend thread with response-channels
pub fn new_response_routine(c: Arc<Mutex<TcpStream>>, config: &Config) {
    let resp_end = ResponseEnd::new(c, config);

    let (resp_tx, resp_rx) = mpsc::sync_channel(config.max_concurrent_streams);

    RESP_TX.set(resp_tx);

    std::thread::Builder::new()
        .name(String::from("pajamax-r")) // response routine
        .spawn(move || response_routine(resp_end, resp_rx))
        .unwrap();
}

// dispatch the request to req_tx
pub fn dispatch<Req>(
    req_tx: &RequestTx<Req>,
    request: Req,
    stream_id: u32,
    req_data_len: usize,
) -> Result<(), Error> {
    let disp_req = DispatchRequest {
        request,
        stream_id,
        req_data_len,
        resp_tx: RESP_TX.with_borrow(|tx| tx.clone()),
    };

    match req_tx.try_send(disp_req) {
        Ok(_) => Ok(()),
        Err(err) => {
            let status = match err {
                mpsc::TrySendError::Full(_) => Status {
                    code: Code::Unavailable,
                    message: String::from("dispatch channel is full"),
                },
                mpsc::TrySendError::Disconnected(_) => Status {
                    code: Code::Internal,
                    message: String::from("dispatch channel is closed"),
                },
            };
            let response: Response<()> = Err(status);
            local_build_response(stream_id, response, req_data_len)
        }
    }
}

// output thread
fn response_routine(mut resp_end: ResponseEnd, resp_rx: ResponseRx) -> Result<(), Error> {
    loop {
        let resp = match resp_rx.try_recv() {
            Ok(resp) => resp,
            Err(mpsc::TryRecvError::Disconnected) => {
                break Err(Error::ChannelClosed);
            }
            Err(mpsc::TryRecvError::Empty) => {
                resp_end.flush()?;
                resp_rx.recv()? // blocking mode
            }
        };

        resp_end.build_box(resp.stream_id, resp.response, resp.req_data_len)?;
    }
}