Streaming Communication Model

Introduction to using Dubbo Rust to quickly develop services for Client streaming, Server streaming, and Bidirectional streaming models.

This article focuses on the Dubbo Rust Streaming communication pattern. Please refer to the Quick Start to understand the basic usage of Dubbo Rust and view the complete example for this article.

1 Adding Streaming Model Definitions in IDL

The complete Greeter service definition is as follows, which includes a Unary, Client stream, Server stream, and Bidirectional stream model for Dubbo services.

  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 Writing Logic with Streaming Model Definitions

2.1 Writing the 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 dropped
  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.         tokio::spawn(async move {
  119.             while let Some(result) = in_stream.next().await {
  120.                 match result {
  121.                     Ok(v) => {
  122.                         tx.send(Ok(GreeterReply {
  123.                             message: format!("server reply: {:?}", v.name),
  124.                         }))
  125.                         .await
  126.                         .expect("working rx")
  127.                     }
  128.                     Err(err) => {
  129.                         if let Some(io_err) = match_for_io_error(&err) {
  130.                             if io_err.kind() == ErrorKind::BrokenPipe {
  131.                                 eprintln!("\tclient disconnected: broken pipe");
  132.                                 break;
  133.                             }
  134.                         }
  136.                         match tx.send(Err(err)).await {
  137.                             Ok(_) => (),
  138.                             Err(_err) => break,
  139.                         }
  140.                     }
  141.                 }
  142.             }
  143.             println!("\tstream ended");
  144.         });
  146.         let out_stream = ReceiverStream::new(rx);
  148.         Ok(Response::new(
  149.             Box::pin(out_stream) as Self::greetStreamStream
  150.         ))
  151.     }
  152. }
  154. fn match_for_io_error(err_status: &dubbo::status::Status) -> Option<&std::io::Error> {
  155.     let mut err: &(dyn std::error::Error + 'static) = err_status;
  157.     loop {
  158.         if let Some(io_err) = err.downcast_ref::<std::io::Error>() {
  159.             return Some(io_err);
  160.         }
  162.         err = match err.source() {
  163.             Some(err) => err,
  164.             None => return None,
  165.         };
  166.     }
  167. }

2.2 Writing the 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 Running the Example

  1. Build

Run cargo build to compile the server and client.

  1. Run the server

Run ./target/debug/greeter-server to start the server. As configured in the above dubbo.yaml, the server will listen on port 8888 and provide RPC services via the triple protocol:

  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. Run the client to see the Streaming communication effects

Run ./target/debug/greeter-client to execute the client, calling various methods under 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"} })

Feedback

Was this page helpful?

Yes No

Last modified September 30, 2024: Update & Translate Overview Docs (#3040) (d37ebceaea7)