gRPC \u662f\u5f00\u53d1\u4e2d\u5e38\u7528\u7684\u5f00\u6e90\u9ad8\u6027\u80fd\u8fdc\u7a0b\u8fc7\u7a0b\u8c03\u7528\uff08RPC\uff09\u6846\u67b6\uff0ctonic \u662f\u57fa\u4e8e HTTP\/2 \u7684 gRPC \u5b9e\u73b0\uff0c\u4e13\u6ce8\u4e8e\u9ad8\u6027\u80fd\u3001\u4e92\u64cd\u4f5c\u6027\u548c\u7075\u6d3b\u6027\u3002\u8be5\u5e93\u7684\u521b\u5efa\u662f\u4e3a\u4e86\u5bf9 async\/await \u63d0\u4f9b\u4e00\u6d41\u7684\u652f\u6301\uff0c\u5e76\u5145\u5f53\u7528 Rust \u7f16\u5199\u7684\u751f\u4ea7\u7cfb\u7edf\u7684\u6838\u5fc3\u6784\u5efa\u5757\u3002\u4eca\u5929\u6211\u4eec\u804a\u804a\u901a\u8fc7\u4f7f\u7528tonic \u8c03\u7528grpc\u7684\u7684\u5177\u4f53\u8fc7\u7a0b\u3002<\/p>\n
rpc\u7a0b\u5e8f\u4e00\u822c\u5305\u542bserver\u7aef\u548cclient\u7aef\uff0c\u4e3a\u4e86\u65b9\u4fbf\u6211\u4eec\u628a\u4e24\u4e2a\u7a0b\u5e8f\u6253\u5305\u5230\u4e00\u4e2a\u5de5\u7a0b\u91cc\u9762 \u65b0\u5efatonic_sample\u5de5\u7a0b<\/p>\n
cargo new tonic_sample\n\n\n\n<\/code><\/pre>\nCargo.toml \u5982\u4e0b<\/p>\n
[package]\nname = "tonic_sample"\nversion = "0.1.0"\nedition = "2021"\n\n[[bin]] # Bin to run the gRPC server\nname = "stream-server"\npath = "src\/stream_server.rs"\n\n[[bin]] # Bin to run the gRPC client\nname = "stream-client"\npath = "src\/stream_client.rs"\n\n\n[dependencies]\ntokio.workspace = true\ntonic = "0.9"\ntonic-reflection = "0.9.2"\nprost = "0.11"\ntokio-stream = "0.1"\nasync-stream = "0.2"\nserde = { version = "1.0", features = ["derive"] }\nserde_json = "1.0"\nrand = "0.7"\nh2 = { version = "0.3" }\nanyhow = "1.0.75"\nfutures-util = "0.3.28"\n\n[build-dependencies]\ntonic-build = "0.9"\n\n\n\n<\/code><\/pre>\ntonic \u7684\u793a\u4f8b\u4ee3\u7801\u8fd8\u662f\u6bd4\u8f83\u9f50\u5168\u7684\uff0c\u672c\u6b21\u6211\u4eec\u53c2\u8003 tonic \u7684 streaming example<\/a>\u3002<\/p>\n\u9996\u5148\u7f16\u5199 proto \u6587\u4ef6\uff0c\u7528\u6765\u63cf\u8ff0\u62a5\u6587\u3002 proto\/echo.proto<\/p>\n
syntax = "proto3";\n\npackage stream;\n\n\/\/ EchoRequest is the request for echo.\nmessage EchoRequest { string message = 1; }\n\n\/\/ EchoResponse is the response for echo.\nmessage EchoResponse { string message = 1; }\n\n\/\/ Echo is the echo service.\nservice Echo {\n \/\/ UnaryEcho is unary echo.\n rpc UnaryEcho(EchoRequest) returns (EchoResponse) {}\n \/\/ ServerStreamingEcho is server side streaming.\n rpc ServerStreamingEcho(EchoRequest) returns (stream EchoResponse) {}\n \/\/ ClientStreamingEcho is client side streaming.\n rpc ClientStreamingEcho(stream EchoRequest) returns (EchoResponse) {}\n \/\/ BidirectionalStreamingEcho is bidi streaming.\n rpc BidirectionalStreamingEcho(stream EchoRequest)\n returns (stream EchoResponse) {}\n}\n\n\n\n<\/code><\/pre>\n\u6587\u4ef6\u5e76\u4e0d\u590d\u6742\uff0c\u53ea\u6709\u4e24\u4e2a message \u4e00\u4e2a\u8bf7\u6c42\u4e00\u4e2a\u8fd4\u56de\uff0c\u4e4b\u6240\u4ee5\u9009\u62e9\u8fd9\u4e2a\u793a\u4f8b\u662f\u56e0\u4e3a\u8be5\u793a\u4f8b\u5305\u542b\u4e86rpc\u4e2d\u7684\u6d41\u5f0f\u5904\u7406\uff0c\u5305\u6269\u4e86server \u6d41\u3001client \u6d41\u4ee5\u53ca\u53cc\u5411\u6d41\u7684\u64cd\u4f5c\u3002 \u7f16\u8f91build.rs<\/a> \u6587\u4ef6<\/p>\nuse std::{env, path::PathBuf};\n\nfn main() -> Result<(), Box<dyn std::error::Error>> {\n tonic_build::compile_protos("proto\/echo.proto")?;\n Ok(())\n}\n\n\n\n<\/code><\/pre>\n\u8be5\u6587\u4ef6\u7528\u6765\u901a\u8fc7 tonic-build \u751f\u6210 grpc \u7684 rust \u57fa\u7840\u4ee3\u7801<\/p>\n
\u5b8c\u6210\u4e0a\u8ff0\u5de5\u4f5c\u540e\u5c31\u53ef\u4ee5\u6784\u5efa server \u548c client \u4ee3\u7801\u4e86<\/p>\n
stream_server.rs<\/p>\n
pub mod pb {\n tonic::include_proto!("stream");\n}\n\nuse anyhow::Result;\nuse futures_util::FutureExt;\nuse pb::{EchoRequest, EchoResponse};\nuse std::{\n error::Error,\n io::ErrorKind,\n net::{SocketAddr, ToSocketAddrs},\n pin::Pin,\n thread,\n time::Duration,\n};\nuse tokio::{\n net::TcpListener,\n sync::{\n mpsc,\n oneshot::{self, Receiver, Sender},\n Mutex,\n },\n task::{self, JoinHandle},\n};\nuse tokio_stream::{\n wrappers::{ReceiverStream, TcpListenerStream},\n Stream, StreamExt,\n};\nuse tonic::{transport::Server, Request, Response, Status, Streaming};\ntype EchoResult<T> = Result<Response<T>, Status>;\ntype ResponseStream = Pin<Box<dyn Stream<Item = Result<EchoResponse, Status>> + Send>>;\n\nfn match_for_io_error(err_status: &Status) -> Option<&std::io::Error> {\n let mut err: &(dyn Error + 'static) = err_status;\n\n loop {\n if let Some(io_err) = err.downcast_ref::<std::io::Error>() {\n return Some(io_err);\n }\n\n \/\/ h2::Error do not expose std::io::Error with `source()`\n \/\/ https:\/\/github.com\/hyperium\/h2\/pull\/462\n if let Some(h2_err) = err.downcast_ref::<h2::Error>() {\n if let Some(io_err) = h2_err.get_io() {\n return Some(io_err);\n }\n }\n\n err = match err.source() {\n Some(err) => err,\n None => return None,\n };\n }\n}\n\n#[derive(Debug)]\npub struct EchoServer {}\n\n#[tonic::async_trait]\nimpl pb::echo_server::Echo for EchoServer {\n async fn unary_echo(&self, req: Request<EchoRequest>) -> EchoResult<EchoResponse> {\n let req_str = req.into_inner().message;\n\n let response = EchoResponse { message: req_str };\n Ok(Response::new(response))\n }\n\n type ServerStreamingEchoStream = ResponseStream;\n\n async fn server_streaming_echo(\n &self,\n req: Request<EchoRequest>,\n ) -> EchoResult<Self::ServerStreamingEchoStream> {\n println!("EchoServer::server_streaming_echo");\n println!("\/tclient connected from: {:?}", req.remote_addr());\n\n \/\/ creating infinite stream with requested message\n let repeat = std::iter::repeat(EchoResponse {\n message: req.into_inner().message,\n });\n let mut stream = Box::pin(tokio_stream::iter(repeat).throttle(Duration::from_millis(200)));\n\n let (tx, rx) = mpsc::channel(128);\n tokio::spawn(async move {\n while let Some(item) = stream.next().await {\n match tx.send(Result::<_, Status>::Ok(item)).await {\n Ok(_) => {\n \/\/ item (server response) was queued to be send to client\n }\n Err(_item) => {\n \/\/ output_stream was build from rx and both are dropped\n break;\n }\n }\n }\n println!("\/tclient disconnected");\n });\n\n let output_stream = ReceiverStream::new(rx);\n Ok(Response::new(\n Box::pin(output_stream) as Self::ServerStreamingEchoStream\n ))\n }\n\n async fn client_streaming_echo(\n &self,\n _: Request<Streaming<EchoRequest>>,\n ) -> EchoResult<EchoResponse> {\n Err(Status::unimplemented("not implemented"))\n }\n\n type BidirectionalStreamingEchoStream = ResponseStream;\n\n async fn bidirectional_streaming_echo(\n &self,\n req: Request<Streaming<EchoRequest>>,\n ) -> EchoResult<Self::BidirectionalStreamingEchoStream> {\n println!("EchoServer::bidirectional_streaming_echo");\n\n let mut in_stream = req.into_inner();\n let (tx, rx) = mpsc::channel(128);\n\n tokio::spawn(async move {\n while let Some(result) = in_stream.next().await {\n match result {\n Ok(v) => tx\n .send(Ok(EchoResponse { message: v.message }))\n .await\n .expect("working rx"),\n Err(err) => {\n if let Some(io_err) = match_for_io_error(&err) {\n if io_err.kind() == ErrorKind::BrokenPipe {\n eprintln!("\/tclient disconnected: broken pipe");\n break;\n }\n }\n\n match tx.send(Err(err)).await {\n Ok(_) => (),\n Err(_err) => break, \/\/ response was droped\n }\n }\n }\n }\n println!("\/tstream ended");\n });\n\n \/\/ echo just write the same data that was received\n let out_stream = ReceiverStream::new(rx);\n\n Ok(Response::new(\n Box::pin(out_stream) as Self::BidirectionalStreamingEchoStream\n ))\n }\n}\n\n#[tokio::main]\nasync fn main() -> Result<(), Box<dyn std::error::Error>> {\n \/\/ \u57fa\u7840server\n let server = EchoServer {};\n Server::builder()\n .add_service(pb::echo_server::EchoServer::new(server))\n .serve("0.0.0.0:50051".to_socket_addrs().unwrap().next().unwrap())\n .await\n .unwrap();\n Ok(())\n}\n\n\n\n\n<\/code><\/pre>\nserver \u7aef\u7684\u4ee3\u7801\u8fd8\u662f\u6bd4\u8f83\u6e05\u6670\u7684\uff0c\u9996\u5148\u901a\u8fc7 tonic::include\/_proto! \u5b8f\u5f15\u5165grpc\u5b9a\u4e49\uff0c\u53c2\u6570\u662f proto \u6587\u4ef6\u4e2d\u5b9a\u4e49\u7684 package \u3002\u6211\u4eec\u91cd\u70b9\u8bf4\u8bf4 server\/_streaming\/_echo function \u3002\u8fd9\u4e2afunction \u7684\u5904\u7406\u6d41\u7a0b\u660e\u767d\u4e86\uff0c\u5176\u4ed6\u7684\u6d41\u5f0f\u5904\u7406\u5927\u540c\u5c0f\u5f02\u3002\u9996\u5148 \u901a\u8fc7std::iter::repeat function \u5b9a\u4e49\u4e00\u4e2a\u8fed\u4ee3\u5668\uff1b\u7136\u540e\u6784\u5efa tokio\/_stream \u5728\u672c\u793a\u4f8b\u4e2d \u6bcf 200\u6beb\u79d2\u4ea7\u751f\u4e00\u4e2a repeat\uff1b\u6700\u540e\u6784\u5efa\u4e00\u4e2a channel \uff0ctx \u7528\u6765\u53d1\u9001\u4ecestream\u4e2d\u83b7\u53d6\u7684\u5185\u5bb9\u592a\uff0crx \u5c01\u88c5\u5230response \u4e2d\u8fd4\u56de\u3002 \u6700\u540e main \u51fd\u6570 \u62c9\u8d77\u670d\u52a1\u3002<\/p>\n
client \u4ee3\u7801\u5982\u4e0b<\/p>\n
pub mod pb {\n tonic::include_proto!("stream");\n}\n\nuse std::time::Duration;\nuse tokio_stream::{Stream, StreamExt};\nuse tonic::transport::Channel;\n\nuse pb::{echo_client::EchoClient, EchoRequest};\n\nfn echo_requests_iter() -> impl Stream<Item = EchoRequest> {\n tokio_stream::iter(1..usize::MAX).map(|i| EchoRequest {\n message: format!("msg {:02}", i),\n })\n}\n\nasync fn unary_echo(client: &mut EchoClient<Channel>, num: usize) {\n for i in 0..num {\n let req = tonic::Request::new(EchoRequest {\n message: "msg".to_string() + &i.to_string(),\n });\n let resp = client.unary_echo(req).await.unwrap();\n println!("resp:{}", resp.into_inner().message);\n }\n}\n\nasync fn streaming_echo(client: &mut EchoClient<Channel>, num: usize) {\n let stream = client\n .server_streaming_echo(EchoRequest {\n message: "foo".into(),\n })\n .await\n .unwrap()\n .into_inner();\n\n \/\/ stream is infinite - take just 5 elements and then disconnect\n let mut stream = stream.take(num);\n while let Some(item) = stream.next().await {\n println!("\/treceived: {}", item.unwrap().message);\n }\n \/\/ stream is droped here and the disconnect info is send to server\n}\n\nasync fn bidirectional_streaming_echo(client: &mut EchoClient<Channel>, num: usize) {\n let in_stream = echo_requests_iter().take(num);\n\n let response = client\n .bidirectional_streaming_echo(in_stream)\n .await\n .unwrap();\n\n let mut resp_stream = response.into_inner();\n\n while let Some(received) = resp_stream.next().await {\n let received = received.unwrap();\n println!("\/treceived message: `{}`", received.message);\n }\n}\n\nasync fn bidirectional_streaming_echo_throttle(client: &mut EchoClient<Channel>, dur: Duration) {\n let in_stream = echo_requests_iter().throttle(dur);\n\n let response = client\n .bidirectional_streaming_echo(in_stream)\n .await\n .unwrap();\n\n let mut resp_stream = response.into_inner();\n\n while let Some(received) = resp_stream.next().await {\n let received = received.unwrap();\n println!("\/treceived message: `{}`", received.message);\n }\n}\n\n#[tokio::main]\nasync fn main() -> Result<(), Box<dyn std::error::Error>> {\n let mut client = EchoClient::connect("http:\/\/127.0.0.1:50051").await.unwrap();\n println!("Unary echo:");\n unary_echo(&mut client, 10).await;\n tokio::time::sleep(Duration::from_secs(1)).await;\n\n println!("Streaming echo:");\n streaming_echo(&mut client, 5).await;\n tokio::time::sleep(Duration::from_secs(1)).await; \/\/do not mess server println functions\n\n \/\/ Echo stream that sends 17 requests then graceful end that connection\n println!("\/r\/nBidirectional stream echo:");\n bidirectional_streaming_echo(&mut client, 17).await;\n\n \/\/ Echo stream that sends up to `usize::MAX` requests. One request each 2s.\n \/\/ Exiting client with CTRL+C demonstrate how to distinguish broken pipe from\n \/\/ graceful client disconnection (above example) on the server side.\n println!("\/r\/nBidirectional stream echo (kill client with CTLR+C):");\n bidirectional_streaming_echo_throttle(&mut client, Duration::from_secs(2)).await;\n\n Ok(())\n}\n\n\n\n\n<\/code><\/pre>\n\u6d4b\u8bd5\u4e00\u4e0b,\u5206\u522b\u8fd0\u884c server \u548c client<\/p>\n
cargo run --bin stream-server\ncargo run --bin stream-client\n\n\n\n<\/code><\/pre>\n\u5728\u5f00\u53d1\u4e2d\uff0c\u6211\u4eec\u901a\u5e38\u4e0d\u4f1a\u518d client \u548c server\u90fd\u5f00\u53d1\u597d\u7684\u60c5\u51b5\u4e0b\u624d\u5f00\u59cb\u6d4b\u8bd5\u3002\u901a\u5e38\u5728\u5f00\u53d1server \u7aef\u7684\u65f6\u5019\u91c7\u7528 grpcurl \u5de5\u5177\u8fdb\u884c\u6d4b\u8bd5\u5de5\u4f5c<\/p>\n
grpcurl -import-path .\/proto -proto echo.proto list\ngrpcurl -import-path .\/proto -proto echo.proto describe stream.Echo\ngrpcurl -plaintext -import-path .\/proto -proto echo.proto -d '{"message":"1234"}' 127.0.0.1:50051 stream.Echo\/UnaryEcho\n\n\n\n<\/code><\/pre>\n\u6b64\u65f6\uff0c\u5982\u679c\u6211\u4eec\u4e0d\u6307\u5b9a -import-path \u53c2\u6570\uff0c\u6267\u884c\u5982\u4e0b\u547d\u4ee4<\/p>\n
grpcurl -plaintext 127.0.0.1:50051 list\n\n\n\n<\/code><\/pre>\n\u4f1a\u51fa\u73b0\u5982\u4e0b\u62a5\u9519\u4fe1\u606f<\/p>\n
Failed to list services: server does not support the reflection API\n\n\n\n<\/code><\/pre>\n\u8ba9\u670d\u52a1\u7aef\u7a0b\u5e8f\u652f\u6301 reflection API<\/h2>\n
\u9996\u5148\u6539\u9020build.rs<\/a><\/p>\nuse std::{env, path::PathBuf};\n\nfn main() -> Result<(), Box<dyn std::error::Error>> {\n let out_dir = PathBuf::from(env::var("OUT_DIR").unwrap());\n tonic_build::configure()\n .file_descriptor_set_path(out_dir.join("stream_descriptor.bin"))\n .compile(&["proto\/echo.proto"], &["proto"])\n .unwrap();\n Ok(())\n}\n\n\n\n<\/code><\/pre>\nfile\/_descriptor\/_set_path \u751f\u6210\u4e00\u4e2a\u6587\u4ef6\uff0c\u5176\u4e2d\u5305\u542b\u4e3a\u534f\u8bae\u7f13\u51b2\u6a21\u5757\u7f16\u7801\u7684 prost_types::FileDescriptorSet<\/code> \u6587\u4ef6\u3002\u8fd9\u662f\u5b9e\u73b0 gRPC \u670d\u52a1\u5668\u53cd\u5c04\u6240\u5fc5\u9700\u7684\u3002<\/p>\n\u63a5\u4e0b\u6765\u6539\u9020\u4e00\u4e0b stream-server.rs<\/a>\uff0c\u6d89\u53ca\u4e24\u5904\u66f4\u6539\u3002<\/p>\n\u65b0\u589e STREAM\/_DESCRIPTOR\/_SET \u5e38\u91cf<\/p>\n
pub mod pb {\n tonic::include_proto!("stream");\n pub const STREAM_DESCRIPTOR_SET: &[u8] =\n tonic::include_file_descriptor_set!("stream_descriptor");\n}\n\n\n\n<\/code><\/pre>\n\u4fee\u6539main\u51fd\u6570<\/p>\n
#[tokio::main]\nasync fn main() -> Result<(), Box<dyn std::error::Error>> {\n \/\/ \u57fa\u7840server\n \/\/ let server = EchoServer {};\n \/\/ Server::builder()\n \/\/ .add_service(pb::echo_server::EchoServer::new(server))\n \/\/ .serve("0.0.0.0:50051".to_socket_addrs().unwrap().next().unwrap())\n \/\/ .await\n \/\/ .unwrap();\n\n \/\/ tonic_reflection \n let service = tonic_reflection::server::Builder::configure()\n .register_encoded_file_descriptor_set(pb::STREAM_DESCRIPTOR_SET)\n .with_service_name("stream.Echo")\n .build()\n .unwrap();\n\n let addr = "0.0.0.0:50051".parse().unwrap();\n\n let server = EchoServer {};\n\n Server::builder()\n .add_service(service)\n .add_service(pb::echo_server::EchoServer::new(server))\n .serve(addr)\n .await?;\n Ok(())\n}\n\n\n\n<\/code><\/pre>\nregister\/_encoded\/_file\/_descriptor\/_set \u5c06\u5305\u542b\u7f16\u7801\u7684 prost_types::FileDescriptorSet<\/code> \u7684 byte slice \u6ce8\u518c\u5230 gRPC Reflection \u670d\u52a1\u751f\u6210\u5668\u6ce8\u518c\u3002<\/p>\n\u518d\u6b21\u6d4b\u8bd5<\/p>\n
grpcurl -plaintext 127.0.0.1:50051 list\ngrpcurl -plaintext 127.0.0.1:50051 describe stream.Echo\n\n\n\n<\/code><\/pre>\n\u8fd4\u56de\u6b63\u786e\u7ed3\u679c\u3002<\/p>\n