tonic-rpc 0.2.1

RPC library that uses native Rust types and is built on tonic.
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

#![cfg(feature = "json")]

use std::time::Duration;

use tokio::{net::TcpListener, sync::mpsc};
use tokio_stream::wrappers::{ReceiverStream, TcpListenerStream};
use tonic::transport::Server;
use tonic_rpc::tonic_rpc;

mod util;

#[tonic_rpc(json)]
trait Counter {
    #[server_streaming]
    fn count(start: i32) -> i32;
    #[server_streaming]
    fn count_n(start: i32, values_count: usize) -> i32;
}

type State = ();

#[tonic::async_trait]
impl counter_server::Counter for State {
    type CountStream = ReceiverStream<Result<i32, tonic::Status>>;

    async fn count(
        &self,
        request: tonic::Request<i32>,
    ) -> Result<tonic::Response<Self::CountStream>, tonic::Status> {
        let mut x = request.into_inner();
        let (tx, rx) = mpsc::channel(100);
        tokio::spawn(async move {
            loop {
                tx.send(Ok(x)).await.unwrap();
                x += 1;
                tokio::time::sleep(Duration::from_millis(1)).await;
            }
        });
        Ok(tonic::Response::new(ReceiverStream::new(rx)))
    }

    type CountNStream = ReceiverStream<Result<i32, tonic::Status>>;

    async fn count_n(
        &self,
        request: tonic::Request<(i32, usize)>,
    ) -> Result<tonic::Response<Self::CountNStream>, tonic::Status> {
        let (start, count) = request.into_inner();
        let (tx, rx) = mpsc::channel(1);
        tokio::spawn(async move {
            for i in 0..count {
                tx.send(Ok(start + (i as i32))).await.unwrap();
            }
        });
        Ok(tonic::Response::new(ReceiverStream::new(rx)))
    }
}

pub async fn run_server() -> u16 {
    let listener = TcpListener::bind("[::1]:0").await.unwrap();
    let port = listener.local_addr().unwrap().port();

    tokio::spawn(async move {
        Server::builder()
            .add_service(counter_server::CounterServer::new(()))
            .serve_with_incoming(TcpListenerStream::new(listener))
            .await
            .unwrap();
    });
    port
}

#[tokio::test]
async fn test_server_streaming() {
    let addr = util::run_server(counter_server::CounterServer::new(())).await;
    let mut client = counter_client::CounterClient::connect(addr)
        .await
        .expect("Failed to connect");

    let request = tonic::Request::new(42_i32);
    let mut responses = client
        .count(request)
        .await
        .expect("Failed to send request")
        .into_inner();
    let first = responses.message().await.unwrap().unwrap();
    assert_eq!(42, first);
    let second = responses.message().await.unwrap().unwrap();
    assert_eq!(43, second);
    let third = responses.message().await.unwrap().unwrap();
    assert_eq!(44, third);
}

#[tokio::test]
async fn test_server_stream_ends() {
    let addr = util::run_server(counter_server::CounterServer::new(())).await;
    let mut client = counter_client::CounterClient::connect(addr)
        .await
        .expect("Failed to connect");

    let request = (100_i32, 3_usize);
    let mut responses = client
        .count_n(request)
        .await
        .expect("Failed to send request")
        .into_inner();
    for _ in 0..3 {
        assert!(responses.message().await.unwrap().is_some());
    }
    assert_eq!(None, responses.message().await.unwrap());
}