1 use std::time::Duration;
2
3 use tokio::{net::TcpListener, sync::oneshot};
4 use tokio_stream::StreamExt;
5
6 use integration_tests::pb::{
7 test1_client, test1_server, test_client, test_server, Input, Input1, Output, Output1,
8 };
9 use tonic::codegen::BoxStream;
10 use tonic::service::RoutesBuilder;
11 use tonic::transport::server::TcpIncoming;
12 use tonic::{
13 transport::{Endpoint, Server},
14 Request, Response, Status,
15 };
16
17 #[tokio::test]
multiple_service_using_routes_builder()18 async fn multiple_service_using_routes_builder() {
19 struct Svc1;
20
21 #[tonic::async_trait]
22 impl test_server::Test for Svc1 {
23 async fn unary_call(&self, _req: Request<Input>) -> Result<Response<Output>, Status> {
24 Ok(Response::new(Output {}))
25 }
26 }
27
28 struct Svc2;
29
30 #[tonic::async_trait]
31 impl test1_server::Test1 for Svc2 {
32 async fn unary_call(&self, request: Request<Input1>) -> Result<Response<Output1>, Status> {
33 Ok(Response::new(Output1 {
34 buf: request.into_inner().buf,
35 }))
36 }
37
38 type StreamCallStream = BoxStream<Output1>;
39
40 async fn stream_call(
41 &self,
42 request: Request<Input1>,
43 ) -> Result<Response<Self::StreamCallStream>, Status> {
44 let output = Output1 {
45 buf: request.into_inner().buf,
46 };
47 let stream = tokio_stream::once(Ok(output));
48
49 Ok(Response::new(Box::pin(stream)))
50 }
51 }
52
53 let svc1 = test_server::TestServer::new(Svc1);
54 let svc2 = test1_server::Test1Server::new(Svc2);
55
56 let (tx, rx) = oneshot::channel::<()>();
57 let mut routes_builder = RoutesBuilder::default();
58 routes_builder.add_service(svc1).add_service(svc2);
59
60 let listener = TcpListener::bind("127.0.0.1:0").await.unwrap();
61 let addr = listener.local_addr().unwrap();
62 let incoming = TcpIncoming::from(listener).with_nodelay(Some(true));
63
64 let jh = tokio::spawn(async move {
65 Server::builder()
66 .add_routes(routes_builder.routes())
67 .serve_with_incoming_shutdown(incoming, async { drop(rx.await) })
68 .await
69 .unwrap();
70 });
71
72 tokio::time::sleep(Duration::from_millis(100)).await;
73
74 let channel = Endpoint::from_shared(format!("http://{addr}"))
75 .unwrap()
76 .connect()
77 .await
78 .unwrap();
79
80 let mut client1 = test_client::TestClient::new(channel.clone());
81 let mut client2 = test1_client::Test1Client::new(channel);
82
83 client1.unary_call(Input {}).await.unwrap();
84
85 let resp2 = client2
86 .unary_call(Input1 {
87 buf: b"hello".to_vec(),
88 })
89 .await
90 .unwrap()
91 .into_inner();
92 assert_eq!(&resp2.buf, b"hello");
93 let mut stream_response = client2
94 .stream_call(Input1 {
95 buf: b"world".to_vec(),
96 })
97 .await
98 .unwrap()
99 .into_inner();
100 let first = match stream_response.next().await {
101 Some(Ok(first)) => first,
102 _ => panic!("expected one non-error item in the stream call response"),
103 };
104
105 assert_eq!(&first.buf, b"world");
106 assert!(stream_response.next().await.is_none());
107
108 tx.send(()).unwrap();
109
110 jh.await.unwrap();
111 }
112