Streaming 通信模型

介绍使用 Dubbo Rust 快速开发 Client streaming、Server streaming、Bidirectional streaming 模型的服务。

本文重点讲解 Dubbo Rust Streaming 通信模式，请先查看 Quick Start 了解 Dubbo Rust 基本使用，在此查看本文的完整示例。

1 IDL 中增加 Streaming 模型定义

完整 Greeter 服务定义如下，包含一个 Unary、Client stream、Server stream、Bidirectional stream 模型的 Dubbo 服务。

  1. // ./proto/greeter.proto
  2. syntax = "proto3";
  4. option java_multiple_files = true;
  6. package org.apache.dubbo.sample.tri;
  9. // The request message containing the user's name.
  10. message GreeterRequest {
  11.   string name = 1;
  12. }
  14. // The response message containing the greetings
  15. message GreeterReply {
  16.   string message = 1;
  17. }
  19. service Greeter{
  21.   // unary
  22.   rpc greet(GreeterRequest) returns (GreeterReply);
  24.   // clientStream
  25.   rpc greetClientStream(stream GreeterRequest) returns (GreeterReply);
  27.   // serverStream
  28.   rpc greetServerStream(GreeterRequest) returns (stream GreeterReply);
  30.   // bi streaming
  31.   rpc greetStream(stream GreeterRequest) returns (stream GreeterReply);
  33. }

2 使用 Streaming 模型定义编写逻辑

2.1 编写 Streaming Server

  1. // ./src/greeter/server.rs
  2. pub mod protos {
  3.     include!(concat!(env!("OUT_DIR"), "/org.apache.dubbo.sample.tri.rs"));
  4. }
  6. use futures_util::StreamExt;
  7. use protos::{
  8.     greeter_server::{register_server, Greeter},
  9.     GreeterReply, GreeterRequest,
  10. };
  12. use std::{io::ErrorKind, pin::Pin};
  14. use async_trait::async_trait;
  15. use futures_util::Stream;
  16. use tokio::sync::mpsc;
  17. use tokio_stream::wrappers::ReceiverStream;
  19. use dubbo_config::RootConfig;
  20. use dubbo::{codegen::*, Dubbo};
  22. type ResponseStream =
  23.     Pin<Box<dyn Stream<Item = Result<GreeterReply, dubbo::status::Status>> + Send>>;
  25. #[tokio::main]
  26. async fn main() {
  27.     register_server(GreeterServerImpl {
  28.         name: "greeter".to_string(),
  29.     });
  31.     // Dubbo::new().start().await;
  32.     Dubbo::new()
  33.         .with_config({
  34.             let r = RootConfig::new();
  35.             match r.load() {
  36.                 Ok(config) => config,
  37.                 Err(_err) => panic!("err: {:?}", _err), // response was droped
  38.             }
  39.         })
  40.         .start()
  41.         .await;
  42. }
  44. #[allow(dead_code)]
  45. #[derive(Default, Clone)]
  46. struct GreeterServerImpl {
  47.     name: String,
  48. }
  50. // #[async_trait]
  51. #[async_trait]
  52. impl Greeter for GreeterServerImpl {
  53.     async fn greet(
  54.         &self,
  55.         request: Request<GreeterRequest>,
  56.     ) -> Result<Response<GreeterReply>, dubbo::status::Status> {
  57.         println!("GreeterServer::greet {:?}", request.metadata);
  59.         Ok(Response::new(GreeterReply {
  60.             message: "hello, dubbo-rust".to_string(),
  61.         }))
  62.     }
  64.     async fn greet_client_stream(
  65.         &self,
  66.         request: Request<Decoding<GreeterRequest>>,
  67.     ) -> Result<Response<GreeterReply>, dubbo::status::Status> {
  68.         let mut s = request.into_inner();
  69.         loop {
  70.             let result = s.next().await;
  71.             match result {
  72.                 Some(Ok(val)) => println!("result: {:?}", val),
  73.                 Some(Err(val)) => println!("err: {:?}", val),
  74.                 None => break,
  75.             }
  76.         }
  77.         Ok(Response::new(GreeterReply {
  78.             message: "hello client streaming".to_string(),
  79.         }))
  80.     }
  82.     type greetServerStreamStream = ResponseStream;
  83.     async fn greet_server_stream(
  84.         &self,
  85.         request: Request<GreeterRequest>,
  86.     ) -> Result<Response<Self::greetServerStreamStream>, dubbo::status::Status> {
  87.         println!("greet_server_stream: {:?}", request.into_inner());
  89.         let data = vec![
  90.             Result::<_, dubbo::status::Status>::Ok(GreeterReply {
  91.                 message: "msg1 from server".to_string(),
  92.             }),
  93.             Result::<_, dubbo::status::Status>::Ok(GreeterReply {
  94.                 message: "msg2 from server".to_string(),
  95.             }),
  96.             Result::<_, dubbo::status::Status>::Ok(GreeterReply {
  97.                 message: "msg3 from server".to_string(),
  98.             }),
  99.         ];
  100.         let resp = futures_util::stream::iter(data);
  102.         Ok(Response::new(Box::pin(resp)))
  103.     }
  105.     type greetStreamStream = ResponseStream;
  106.     async fn greet_stream(
  107.         &self,
  108.         request: Request<Decoding<GreeterRequest>>,
  109.     ) -> Result<Response<Self::greetStreamStream>, dubbo::status::Status> {
  110.         println!(
  111.             "GreeterServer::greet_stream, grpc header: {:?}",
  112.             request.metadata
  113.         );
  115.         let mut in_stream = request.into_inner();
  116.         let (tx, rx) = mpsc::channel(128);
  118.         // this spawn here is required if you want to handle connection error.
  119.         // If we just map `in_stream` and write it back as `out_stream` the `out_stream`
  120.         // will be drooped when connection error occurs and error will never be propagated
  121.         // to mapped version of `in_stream`.
  122.         tokio::spawn(async move {
  123.             while let Some(result) = in_stream.next().await {
  124.                 match result {
  125.                     Ok(v) => {
  126.                         // if v.name.starts_with("msg2") {
  127.                         //     tx.send(Err(dubbo::status::Status::internal(format!("err: args is invalid, {:?}", v.name))
  128.                         //     )).await.expect("working rx");
  129.                         //     continue;
  130.                         // }
  131.                         tx.send(Ok(GreeterReply {
  132.                             message: format!("server reply: {:?}", v.name),
  133.                         }))
  134.                         .await
  135.                         .expect("working rx")
  136.                     }
  137.                     Err(err) => {
  138.                         if let Some(io_err) = match_for_io_error(&err) {
  139.                             if io_err.kind() == ErrorKind::BrokenPipe {
  140.                                 // here you can handle special case when client
  141.                                 // disconnected in unexpected way
  142.                                 eprintln!("\tclient disconnected: broken pipe");
  143.                                 break;
  144.                             }
  145.                         }
  147.                         match tx.send(Err(err)).await {
  148.                             Ok(_) => (),
  149.                             Err(_err) => break, // response was droped
  150.                         }
  151.                     }
  152.                 }
  153.             }
  154.             println!("\tstream ended");
  155.         });
  157.         // echo just write the same data that was received
  158.         let out_stream = ReceiverStream::new(rx);
  160.         Ok(Response::new(
  161.             Box::pin(out_stream) as Self::greetStreamStream
  162.         ))
  163.     }
  164. }
  166. fn match_for_io_error(err_status: &dubbo::status::Status) -> Option<&std::io::Error> {
  167.     let mut err: &(dyn std::error::Error + 'static) = err_status;
  169.     loop {
  170.         if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
  171.             return Some(io_err);
  172.         }
  174.         err = match err.source() {
  175.             Some(err) => err,
  176.             None => return None,
  177.         };
  178.     }
  179. }

2.2 编写 Streaming Client

  1. // ./src/greeter/client.rs
  2. pub mod protos {
  3.     include!(concat!(env!("OUT_DIR"), "/org.apache.dubbo.sample.tri.rs"));
  4. }
  6. use dubbo::codegen::*;
  7. use futures_util::StreamExt;
  8. use protos::{greeter_client::GreeterClient, GreeterRequest};
  10. #[tokio::main]
  11. async fn main() {
  12.     let mut cli = GreeterClient::new().with_uri("http://127.0.0.1:8888".to_string());
  14.     println!("# unary call");
  15.     let resp = cli
  16.         .greet(Request::new(GreeterRequest {
  17.             name: "message from client".to_string(),
  18.         }))
  19.         .await;
  20.     let resp = match resp {
  21.         Ok(resp) => resp,
  22.         Err(err) => return println!("{:?}", err),
  23.     };
  24.     let (_parts, body) = resp.into_parts();
  25.     println!("Response: {:?}", body);
  27.     println!("# client stream");
  28.     let data = vec![
  29.         GreeterRequest {
  30.             name: "msg1 from client streaming".to_string(),
  31.         },
  32.         GreeterRequest {
  33.             name: "msg2 from client streaming".to_string(),
  34.         },
  35.         GreeterRequest {
  36.             name: "msg3 from client streaming".to_string(),
  37.         },
  38.     ];
  39.     let req = futures_util::stream::iter(data);
  40.     let resp = cli.greet_client_stream(req).await;
  41.     let client_streaming_resp = match resp {
  42.         Ok(resp) => resp,
  43.         Err(err) => return println!("{:?}", err),
  44.     };
  45.     let (_parts, resp_body) = client_streaming_resp.into_parts();
  46.     println!("client streaming, Response: {:?}", resp_body);
  48.     println!("# bi stream");
  49.     let data = vec![
  50.         GreeterRequest {
  51.             name: "msg1 from client".to_string(),
  52.         },
  53.         GreeterRequest {
  54.             name: "msg2 from client".to_string(),
  55.         },
  56.         GreeterRequest {
  57.             name: "msg3 from client".to_string(),
  58.         },
  59.     ];
  60.     let req = futures_util::stream::iter(data);
  62.     let bidi_resp = cli.greet_stream(req).await.unwrap();
  64.     let (parts, mut body) = bidi_resp.into_parts();
  65.     println!("parts: {:?}", parts);
  66.     while let Some(item) = body.next().await {
  67.         match item {
  68.             Ok(v) => {
  69.                 println!("reply: {:?}", v);
  70.             }
  71.             Err(err) => {
  72.                 println!("err: {:?}", err);
  73.             }
  74.         }
  75.     }
  76.     let trailer = body.trailer().await.unwrap();
  77.     println!("trailer: {:?}", trailer);
  79.     println!("# server stream");
  80.     let resp = cli
  81.         .greet_server_stream(Request::new(GreeterRequest {
  82.             name: "server streaming req".to_string(),
  83.         }))
  84.         .await
  85.         .unwrap();
  87.     let (parts, mut body) = resp.into_parts();
  88.     println!("parts: {:?}", parts);
  89.     while let Some(item) = body.next().await {
  90.         match item {
  91.             Ok(v) => {
  92.                 println!("reply: {:?}", v);
  93.             }
  94.             Err(err) => {
  95.                 println!("err: {:?}", err);
  96.             }
  97.         }
  98.     }
  99.     let trailer = body.trailer().await.unwrap();
  100.     println!("trailer: {:?}", trailer);
  101. }

3 运行示例

  1. 编译

执行cargo build来编译server和client。

  1. 运行server

执行./target/debug/greeter-server来运行server，如上文dubbo.yaml所配置，server会监听8888端口，并以triple协议提供RPC服务：

  1. $ ./target/debug/greeter-server
  2. 2022-09-28T23:33:28.104577Z  INFO dubbo::framework: url: Some(Url { uri: "triple://0.0.0.0:8888/org.apache.dubbo.sample.tri.Greeter", protocol: "triple", location: "0.0.0.0:8888", ip: "0.0.0.0", port: "8888", service_key: ["org.apache.dubbo.sample.tri.Greeter"], params: {} })
  1. 运行client，可以看到 Streaming 通信效果

执行./target/debug/greeter-client来运行client，调用triple://127.0.0.1:8888/org.apache.dubbo.sample.tri.Greeter下的各种方法：

  1. $ ./target/debug/greeter-client
  2. # unary call
  3. Response: GreeterReply { message: "hello, dubbo-rust" }
  4. # client stream
  5. client streaming, Response: GreeterReply { message: "hello client streaming" }
  6. # bi stream
  7. parts: Metadata { inner: {"content-type": "application/grpc", "date": "Wed, 28 Sep 2022 23:34:20 GMT"} }
  8. reply: GreeterReply { message: "server reply: \"msg1 from client\"" }
  9. reply: GreeterReply { message: "server reply: \"msg2 from client\"" }
  10. reply: GreeterReply { message: "server reply: \"msg3 from client\"" }
  11. trailer: Some(Metadata { inner: {"content-type": "application/grpc", "grpc-status": "0", "grpc-message": "poll trailer successfully.", "grpc-accept-encoding": "gzip,identity"} })
  12. # server stream
  13. parts: Metadata { inner: {"content-type": "application/grpc", "date": "Wed, 28 Sep 2022 23:34:20 GMT"} }
  14. reply: GreeterReply { message: "msg1 from server" }
  15. reply: GreeterReply { message: "msg2 from server" }
  16. reply: GreeterReply { message: "msg3 from server" }
  17. trailer: Some(Metadata { inner: {"content-type": "application/grpc", "grpc-status": "0", "grpc-message": "poll trailer successfully.", "grpc-accept-encoding": "gzip,identity"} })

最后修改 December 16, 2022: Fix check (#1736) (97972c1)