feat(tonic-web): implement grpc <-> grpc-web protocol translation (#455)
tonic-web enables tonic servers to handle requests from grpc-web clients directly, without the need of an external proxy. Co-authored-by: John Hernandez <[email protected]> Co-authored-by: zancas <[email protected]>
This commit is contained in:
Juan Alvarez
co-authored by
John Hernandez
zancas
John Hernandez
zancas
parent
352b0f584b
commit
c309063254
@@ -0,0 +1,167 @@
|
||||
use std::future::Future;
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use tokio::net::TcpListener;
|
||||
use tokio::time::Duration;
|
||||
use tokio::{join, try_join};
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tokio_stream::{self as stream, StreamExt};
|
||||
use tonic::transport::{Channel, Error, Server};
|
||||
use tonic::{Response, Streaming};
|
||||
|
||||
use integration::pb::{test_client::TestClient, test_server::TestServer, Input};
|
||||
use integration::Svc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn smoke_unary() {
|
||||
let (mut c1, mut c2, mut c3, mut c4) = spawn().await.expect("clients");
|
||||
|
||||
let (r1, r2, r3, r4) = try_join!(
|
||||
c1.unary_call(input()),
|
||||
c2.unary_call(input()),
|
||||
c3.unary_call(input()),
|
||||
c4.unary_call(input()),
|
||||
)
|
||||
.expect("responses");
|
||||
|
||||
assert!(meta(&r1) == meta(&r2) && meta(&r2) == meta(&r3) && meta(&r3) == meta(&r4));
|
||||
assert!(data(&r1) == data(&r2) && data(&r2) == data(&r3) && data(&r3) == data(&r4));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn smoke_client_stream() {
|
||||
let (mut c1, mut c2, mut c3, mut c4) = spawn().await.expect("clients");
|
||||
|
||||
let input_stream = || stream::iter(vec![input(), input()]);
|
||||
|
||||
let (r1, r2, r3, r4) = try_join!(
|
||||
c1.client_stream(input_stream()),
|
||||
c2.client_stream(input_stream()),
|
||||
c3.client_stream(input_stream()),
|
||||
c4.client_stream(input_stream()),
|
||||
)
|
||||
.expect("responses");
|
||||
|
||||
assert!(meta(&r1) == meta(&r2) && meta(&r2) == meta(&r3) && meta(&r3) == meta(&r4));
|
||||
assert!(data(&r1) == data(&r2) && data(&r2) == data(&r3) && data(&r3) == data(&r4));
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn smoke_server_stream() {
|
||||
let (mut c1, mut c2, mut c3, mut c4) = spawn().await.expect("clients");
|
||||
|
||||
let (r1, r2, r3, r4) = try_join!(
|
||||
c1.server_stream(input()),
|
||||
c2.server_stream(input()),
|
||||
c3.server_stream(input()),
|
||||
c4.server_stream(input()),
|
||||
)
|
||||
.expect("responses");
|
||||
|
||||
assert!(meta(&r1) == meta(&r2) && meta(&r2) == meta(&r3) && meta(&r3) == meta(&r4));
|
||||
|
||||
let r1 = stream(r1).await;
|
||||
let r2 = stream(r2).await;
|
||||
let r3 = stream(r3).await;
|
||||
let r4 = stream(r4).await;
|
||||
|
||||
assert!(&r1 == &r2 && &r2 == &r3 && &r3 == &r4);
|
||||
}
|
||||
#[tokio::test]
|
||||
async fn smoke_error() {
|
||||
let (mut c1, mut c2, mut c3, mut c4) = spawn().await.expect("clients");
|
||||
|
||||
let boom = Input {
|
||||
id: 1,
|
||||
desc: "boom".to_owned(),
|
||||
};
|
||||
|
||||
let (r1, r2, r3, r4) = join!(
|
||||
c1.unary_call(boom.clone()),
|
||||
c2.unary_call(boom.clone()),
|
||||
c3.unary_call(boom.clone()),
|
||||
c4.unary_call(boom.clone()),
|
||||
);
|
||||
|
||||
let s1 = r1.unwrap_err();
|
||||
let s2 = r2.unwrap_err();
|
||||
let s3 = r3.unwrap_err();
|
||||
let s4 = r4.unwrap_err();
|
||||
|
||||
assert!(status(&s1) == status(&s2) && status(&s2) == status(&s3) && status(&s3) == status(&s4))
|
||||
}
|
||||
|
||||
async fn bind() -> (TcpListener, String) {
|
||||
let addr = SocketAddr::from(([127, 0, 0, 1], 0));
|
||||
let lis = TcpListener::bind(addr).await.expect("listener");
|
||||
let url = format!("http://{}", lis.local_addr().unwrap());
|
||||
|
||||
(lis, url)
|
||||
}
|
||||
|
||||
async fn grpc(accept_h1: bool) -> (impl Future<Output = Result<(), Error>>, String) {
|
||||
let (listener, url) = bind().await;
|
||||
|
||||
let fut = Server::builder()
|
||||
.accept_http1(accept_h1)
|
||||
.add_service(TestServer::new(Svc))
|
||||
.serve_with_incoming(TcpListenerStream::new(listener));
|
||||
|
||||
(fut, url)
|
||||
}
|
||||
|
||||
async fn grpc_web(accept_h1: bool) -> (impl Future<Output = Result<(), Error>>, String) {
|
||||
let (listener, url) = bind().await;
|
||||
|
||||
let svc = tonic_web::config()
|
||||
.allow_origins(vec!["http://foo.com"])
|
||||
.enable(TestServer::new(Svc));
|
||||
|
||||
let fut = Server::builder()
|
||||
.accept_http1(accept_h1)
|
||||
.add_service(svc)
|
||||
.serve_with_incoming(TcpListenerStream::new(listener));
|
||||
|
||||
(fut, url)
|
||||
}
|
||||
|
||||
type Client = TestClient<Channel>;
|
||||
|
||||
async fn spawn() -> Result<(Client, Client, Client, Client), Error> {
|
||||
let ((s1, u1), (s2, u2), (s3, u3), (s4, u4)) =
|
||||
join!(grpc(true), grpc(false), grpc_web(true), grpc_web(false));
|
||||
|
||||
let _ = tokio::spawn(async move { join!(s1, s2, s3, s4) });
|
||||
|
||||
tokio::time::sleep(Duration::from_millis(30)).await;
|
||||
|
||||
try_join!(
|
||||
TestClient::connect(u1),
|
||||
TestClient::connect(u2),
|
||||
TestClient::connect(u3),
|
||||
TestClient::connect(u4)
|
||||
)
|
||||
}
|
||||
|
||||
fn input() -> Input {
|
||||
Input {
|
||||
id: 1,
|
||||
desc: "one".to_owned(),
|
||||
}
|
||||
}
|
||||
|
||||
fn meta<T>(r: &Response<T>) -> String {
|
||||
format!("{:?}", r.metadata())
|
||||
}
|
||||
|
||||
fn data<T>(r: &Response<T>) -> &T {
|
||||
r.get_ref()
|
||||
}
|
||||
|
||||
async fn stream<T>(r: Response<Streaming<T>>) -> Vec<T> {
|
||||
r.into_inner().collect::<Result<Vec<_>, _>>().await.unwrap()
|
||||
}
|
||||
|
||||
fn status(s: &tonic::Status) -> (String, tonic::Code) {
|
||||
(format!("{:?}", s.metadata()), s.code())
|
||||
}
|
||||
@@ -0,0 +1,154 @@
|
||||
use std::net::SocketAddr;
|
||||
|
||||
use bytes::{Buf, BufMut, Bytes, BytesMut};
|
||||
use hyper::http::{header, StatusCode};
|
||||
use hyper::{Body, Client, Method, Request, Uri};
|
||||
use prost::Message;
|
||||
use tokio::net::TcpListener;
|
||||
use tokio_stream::wrappers::TcpListenerStream;
|
||||
use tonic::transport::Server;
|
||||
|
||||
use integration::pb::{test_server::TestServer, Input, Output};
|
||||
use integration::Svc;
|
||||
|
||||
#[tokio::test]
|
||||
async fn binary_request() {
|
||||
let server_url = spawn("http://example.com").await;
|
||||
let client = Client::new();
|
||||
|
||||
let req = build_request(server_url, "grpc-web", "grpc-web");
|
||||
let res = client.request(req).await.unwrap();
|
||||
let content_type = res.headers().get(header::CONTENT_TYPE).unwrap().clone();
|
||||
let content_type = content_type.to_str().unwrap();
|
||||
|
||||
assert_eq!(res.status(), StatusCode::OK);
|
||||
assert_eq!(content_type, "application/grpc-web+proto");
|
||||
|
||||
let (message, trailers) = decode_body(res.into_body(), content_type).await;
|
||||
let expected = Output {
|
||||
id: 1,
|
||||
desc: "one".to_owned(),
|
||||
};
|
||||
|
||||
assert_eq!(message, expected);
|
||||
assert_eq!(&trailers[..], b"grpc-status:0\r\n");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn text_request() {
|
||||
let server_url = spawn("http://example.com").await;
|
||||
let client = Client::new();
|
||||
|
||||
let req = build_request(server_url, "grpc-web-text", "grpc-web-text");
|
||||
let res = client.request(req).await.unwrap();
|
||||
let content_type = res.headers().get(header::CONTENT_TYPE).unwrap().clone();
|
||||
let content_type = content_type.to_str().unwrap();
|
||||
|
||||
assert_eq!(res.status(), StatusCode::OK);
|
||||
assert_eq!(content_type, "application/grpc-web-text+proto");
|
||||
|
||||
let (message, trailers) = decode_body(res.into_body(), content_type).await;
|
||||
let expected = Output {
|
||||
id: 1,
|
||||
desc: "one".to_owned(),
|
||||
};
|
||||
|
||||
assert_eq!(message, expected);
|
||||
assert_eq!(&trailers[..], b"grpc-status:0\r\n");
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn origin_not_allowed() {
|
||||
let server_url = spawn("http://foo.com").await;
|
||||
let client = Client::new();
|
||||
|
||||
let req = build_request(server_url, "grpc-web-text", "grpc-web-text");
|
||||
let res = client.request(req).await.unwrap();
|
||||
|
||||
assert_eq!(res.status(), StatusCode::FORBIDDEN);
|
||||
}
|
||||
|
||||
|
||||
async fn spawn(allowed_origin: &str) -> String {
|
||||
let addr = SocketAddr::from(([127, 0, 0, 1], 0));
|
||||
let listener = TcpListener::bind(addr).await.expect("listener");
|
||||
let url = format!("http://{}", listener.local_addr().unwrap());
|
||||
let listener_stream = TcpListenerStream::new(listener);
|
||||
|
||||
let svc = tonic_web::config()
|
||||
.allow_origins(vec![allowed_origin])
|
||||
.enable(TestServer::new(Svc));
|
||||
|
||||
let _ = tokio::spawn(async move {
|
||||
Server::builder()
|
||||
.accept_http1(true)
|
||||
.add_service(svc)
|
||||
.serve_with_incoming(listener_stream)
|
||||
.await
|
||||
.unwrap()
|
||||
});
|
||||
|
||||
url
|
||||
}
|
||||
|
||||
fn encode_body() -> Bytes {
|
||||
let input = Input {
|
||||
id: 1,
|
||||
desc: "one".to_owned(),
|
||||
};
|
||||
|
||||
let mut buf = BytesMut::with_capacity(1024);
|
||||
buf.reserve(5);
|
||||
unsafe {
|
||||
buf.advance_mut(5);
|
||||
}
|
||||
|
||||
input.encode(&mut buf).unwrap();
|
||||
|
||||
let len = buf.len() - 5;
|
||||
{
|
||||
let mut buf = &mut buf[..5];
|
||||
buf.put_u8(0);
|
||||
buf.put_u32(len as u32);
|
||||
}
|
||||
|
||||
buf.split_to(len + 5).freeze()
|
||||
}
|
||||
|
||||
fn build_request(base_uri: String, content_type: &str, accept: &str) -> Request<Body> {
|
||||
use header::{ACCEPT, CONTENT_TYPE, ORIGIN};
|
||||
|
||||
let request_uri = format!("{}/{}/{}", base_uri, "test.Test", "UnaryCall")
|
||||
.parse::<Uri>()
|
||||
.unwrap();
|
||||
|
||||
let bytes = match content_type {
|
||||
"grpc-web" => encode_body(),
|
||||
"grpc-web-text" => base64::encode(encode_body()).into(),
|
||||
_ => panic!("invalid content type {}", content_type),
|
||||
};
|
||||
|
||||
Request::builder()
|
||||
.method(Method::POST)
|
||||
.header(CONTENT_TYPE, format!("application/{}", content_type))
|
||||
.header(ORIGIN, "http://example.com")
|
||||
.header(ACCEPT, format!("application/{}", accept))
|
||||
.uri(request_uri)
|
||||
.body(Body::from(bytes))
|
||||
.unwrap()
|
||||
}
|
||||
|
||||
async fn decode_body(body: Body, content_type: &str) -> (Output, Bytes) {
|
||||
let mut body = hyper::body::to_bytes(body).await.unwrap();
|
||||
|
||||
if content_type == "application/grpc-web-text+proto" {
|
||||
body = base64::decode(body).unwrap().into()
|
||||
}
|
||||
|
||||
body.advance(1);
|
||||
let len = body.get_u32();
|
||||
let msg = Output::decode(&mut body.split_to(len as usize)).expect("decode");
|
||||
body.advance(5);
|
||||
|
||||
(msg, body)
|
||||
}
|
||||
Reference in New Issue
Block a user