使用Unix套接字连接器通信

介绍使用 Dubbo Rust Triple 协议使用 Unix 套接字连接器实现通信。

本文重点讲解 Dubbo Rust Triple 协议使用Unix 套接，请先查看 Quick Start 了解 Dubbo Rust 基本使用，在此查看本文的完整示例。

1 Triple 使用 Unix 套接字连接器

TripleServer 调整如下。

  1. // dubbo/src/protocol/triple/triple_server.rs
  2. impl TripleServer {
  3.     pub fn new(names: Vec<String>) -> TripleServer {
  4.         Self {
  5.             service_names: names,
  6.             s: DubboServer::new(),
  7.         }
  8.     }
  10.     pub async fn serve(mut self, url: String) {
  11.         {
  12.             let lock = super::TRIPLE_SERVICES.read().unwrap();
  13.             for name in self.service_names.iter() {
  14.                 if lock.get(name).is_none() {
  15.                     tracing::warn!("service ({}) not register", name);
  16.                     continue;
  17.                 }
  18.                 let svc = lock.get(name).unwrap();
  20.                 self.s = self.s.add_service(name.clone(), svc.clone());
  21.             }
  22.         }
  24.         let uri = match http::Uri::from_str(&url) {
  25.             Ok(v) => v,
  26.             Err(err) => {
  27.                 tracing::error!("http uri parse error: {}, url: {:?}", err, &url);
  28.                 return;
  29.             }
  30.         };
  32.         let authority = match uri.authority() {
  33.             Some(v) => v.to_owned(),
  34.             None => {
  35.                 tracing::error!("http authority is none");
  36.                 return;
  37.             }
  38.         };
  40.         self.s
  41.             .with_listener("unix".to_string())
  42.             .serve(
  43.                 authority
  44.                     .to_string()
  45.                     .to_socket_addrs()
  46.                     .unwrap()
  47.                     .next()
  48.                     .unwrap(),
  49.             )
  50.             .await
  51.             .unwrap();
  52.     }
  53. }
  55. }

2 使用 client/connection 使用 Unix 套接字连接器编写逻辑

2.1 编写 Client 端

  1. // examples/echo/src/echo/client.rs
  2. ...
  3. let mut cli = EchoClient::connect("unix://0.0.0.0:8888".to_string());
  4. ...

2.2 编写 connection

  1. // dubbo/src/triple/client/connection.rs
  2. impl Connection {
  3.     pub fn new() -> Self {
  4.         Connection {
  5.             host: hyper::Uri::default(),
  6.             connector: "unix".to_string(),
  7.             builder: Builder::new(),
  8.         }
  9.     }
  11.     pub fn with_connector<C>(mut self, connector: String) -> Self {
  12.         self.connector = connector;
  13.         self
  14.     }
  16.     pub fn with_host(mut self, uri: hyper::Uri) -> Self {
  17.         self.host = uri;
  18.         self
  19.     }
  21.     pub fn with_builder(mut self, builder: Builder) -> Self {
  22.         self.builder = builder;
  23.         self
  24.     }
  25. }

3 运行示例

  1. 编译

执行cargo build来编译server和client。

  1. 运行server

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

  1. $ cargo run --bin echo-server
  2. 2022-12-05T10:40:20.829798Z  INFO dubbo::framework: url: triple://0.0.0.0:8888//grpc.examples.echo.Echo
  3. 2022-12-05T10:40:20.829830Z DEBUG dubbo::framework: service name: grpc.examples.echo.Echo, service_config: ServiceConfig { version: "1.0.0", group: "test", name: "", protocol: "triple", registry: "zookeeper", serializer: "json", protocol_configs: {"triple": ProtocolConfig { ip: "0.0.0.0", port: "8888", name: "triple", params: {} }} }
  4. 2022-12-05T10:40:20.829917Z  INFO dubbo::framework: protocol: "", service url: Url { uri: "triple://0.0.0.0:8888//grpc.examples.echo.Echo", protocol: "triple", location: "0.0.0.0:8888", ip: "0.0.0.0", port: "8888", service_key: ["grpc.examples.echo.Echo"], params: {} }
  5. 2022-12-05T10:40:20.830028Z DEBUG dubbo::triple::transport::listener: get_listener: "unix"
  1. 运行client，可以看到 Unix 通信效果

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

  1. $ cargo run --bin echo-client
  2. # unary call
  3. fake filter: Metadata { inner: {"path": "/grpc.examples.echo.Echo/UnaryEcho", "tri-unit-info": "dubbo-rust/0.1.0", "grpc-accept-encoding": "gzip", "authority": "0.0.0.0:8888", "te": "trailers", "method": "POST", "grpc-encoding": "gzip", "tri-request-time": "1670237034581", "tri-service-group": "cluster", "content-type": "application/grpc+json", "tri-service-version": "dubbo-rust/0.1.0", "scheme": "unix", "user-agent": "dubbo-rust/0.1.0"} }
  4. Response: EchoResponse { message: "hello, dubbo-rust" }
  5. client streaming, Response: EchoResponse { message: "hello client streaming" }
  6. parts: Metadata { inner: {"date": "Mon, 05 Dec 2022 10:43:57 GMT", "content-type": "application/grpc"} }
  7. reply: EchoResponse { message: "server reply: \"msg1 from client\"" }
  8. reply: EchoResponse { message: "server reply: \"msg2 from client\"" }
  9. reply: EchoResponse { message: "server reply: \"msg3 from client\"" }
  10. trailer: Some(Metadata { inner: {"grpc-message": "poll trailer successfully.", "content-type": "application/grpc", "grpc-accept-encoding": "gzip,identity", "grpc-status": "0"} })
  11. parts: Metadata { inner: {"content-type": "application/grpc", "date": "Mon, 05 Dec 2022 10:43:58 GMT"} }
  12. reply: EchoResponse { message: "msg1 from server" }
  13. reply: EchoResponse { message: "msg2 from server" }
  14. reply: EchoResponse { message: "msg3 from server" }
  15. trailer: Some(Metadata { inner: {"grpc-status": "0", "grpc-message": "poll trailer successfully.", "grpc-accept-encoding": "gzip,identity", "content-type": "application/grpc"} })

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