tonic-server-mock 1.0.1

Mock tonic-based gRPC servers that run in memory without networking
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

use std::{convert::Infallible, sync::Arc};
use tokio::{io::DuplexStream, sync::mpsc::Sender};
use tonic::{
    body::BoxBody,
    codegen::{
        http::{Request, Response},
        Service,
    },
    server::NamedService,
    transport::{Channel, Error as TransportError},
};

pub trait Svc:
    Service<Request<BoxBody>, Response = Response<BoxBody>, Error = Infallible>
    + NamedService
    + Clone
    + Send
    + 'static
where
    Self::Future: Send + 'static,
{
}

impl<S> Svc for S
where
    S: Service<Request<BoxBody>, Response = Response<BoxBody>, Error = Infallible>
        + NamedService
        + Clone
        + Send
        + 'static,
    S::Future: Send + 'static,
{
}

pub type ConnectionMockSender = Sender<Result<DuplexStream, TransportError>>;

#[derive(Debug, Clone)]
pub struct EndpointMock {
    connection_sender: Arc<ConnectionMockSender>,
}

impl EndpointMock {
    pub fn new(connection_sender: ConnectionMockSender) -> Self {
        Self {
            connection_sender: Arc::new(connection_sender),
        }
    }

    pub async fn once(self) -> Channel {
        self.connect().await
    }

    pub async fn connect(&self) -> Channel {
        let connection_sender = Arc::clone(&self.connection_sender);

        let client_connector =
            ::tower::service_fn(move |/* mut */ uri: ::tonic::transport::Uri| {
                tracing::info!("connection to {:?}", uri);
                let connection_sender = Arc::clone(&connection_sender);
                async move {
                    let (client_io, server_io) = ::tokio::io::duplex(1024);
                    connection_sender.send(Ok(server_io)).await.unwrap();
                    Ok::<_, ::tonic::transport::Error>(::hyper_util::rt::TokioIo::new(client_io))
                }
            });

        ::tonic::transport::Endpoint::try_from("http://[::1]:50051/pseudo-endpoint")
            .unwrap()
            .connect_with_connector(client_connector)
            .await
            .unwrap()
    }
}

#[macro_export]
macro_rules! mock_server_fn {
    ($vis:vis $fn_name:ident; $($svc:ident),+; $logger:path) => {

        $vis async fn $fn_name(
            $($svc: impl $crate::Svc<Future: Send>),+
        ) -> (impl ::futures::Future<Output = ()>, $crate::EndpointMock) {
            use $logger::{info};

            let (connection_sender, connections_receiver) = ::tokio::sync::mpsc::channel(32);
            let incoming_connections: ::tokio_stream::wrappers::ReceiverStream<
                Result<::tokio::io::DuplexStream, ::tonic::transport::Error>,
            > = ::tokio_stream::wrappers::ReceiverStream::from(connections_receiver);

            let router = ::tonic::transport::Server::builder()$(
                .add_service($svc))+;

            // Spawn server
            let server_future = async move {
                info!("start grpc server");
                router
                    .serve_with_incoming(incoming_connections)
                    .await
                    .unwrap();
                info!("grpc server stopped");
            };

            let connector = $crate::EndpointMock::new(connection_sender);

            (server_future, connector)
        }
    };

    ($fn_name:ident; $($svc:ident),+; $logger:path) => {
        mock_server_fn!(pub(crate) $fn_name; $($svc),+; $logger);
    };

    ($vis:vis $fn_name:ident; $($svc:ident),+) => {
        mock_server_fn!($vis $fn_name; $($svc),+; ::tracing);
    };

    ($fn_name:ident; $($svc:ident),+) => {
        mock_server_fn!(pub(crate) $fn_name; $($svc),+; ::tracing);
    };
}