You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@dubbo.apache.org by li...@apache.org on 2022/07/06 09:41:00 UTC
[dubbo-rust] 24/26: feat(example): Simple service router example.
This is an automated email from the ASF dual-hosted git repository.
liujun pushed a commit to branch poc-transport
in repository https://gitbox.apache.org/repos/asf/dubbo-rust.git
commit 26d144c700eee0956b2cc028912565984535c9ef
Author: even <st...@icloud.com>
AuthorDate: Fri Jun 24 11:06:13 2022 +0800
feat(example): Simple service router example.
---
dubbo-rust-examples/Cargo.toml | 10 ++++
.../src/jsonrpc-basic/addservice.rs | 33 +++++++----
dubbo-rust-examples/src/jsonrpc-basic/client.rs | 16 ++++--
dubbo-rust-examples/src/jsonrpc-basic/server.rs | 26 ++++++---
.../addservice.rs | 33 +++++++----
.../client.rs | 35 ++++++++----
.../mutilservice.rs} | 65 +++++++++++++---------
.../server.rs | 24 ++++++--
8 files changed, 162 insertions(+), 80 deletions(-)
diff --git a/dubbo-rust-examples/Cargo.toml b/dubbo-rust-examples/Cargo.toml
index 3078752..f229f2b 100644
--- a/dubbo-rust-examples/Cargo.toml
+++ b/dubbo-rust-examples/Cargo.toml
@@ -31,3 +31,13 @@ path = "src/jsonrpc-basic/client.rs"
name = "jsonrpc-basic-server"
path = "src/jsonrpc-basic/server.rs"
+#jsonrpc-with-router
+
+[[bin]]
+name = "jsonrpc-with-router-client"
+path = "src/jsonrpc-with-router/client.rs"
+
+
+[[bin]]
+name = "jsonrpc-with-router-server"
+path = "src/jsonrpc-with-router/server.rs"
\ No newline at end of file
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs b/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
index 91f554b..82b35b2 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
+++ b/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
@@ -35,11 +35,11 @@ pub type AddResp = i32;
#[async_trait]
pub trait AddService {
- async fn add(&mut self, req: AddReq) -> Result<AddResp, StdError>;
+ async fn add(&self, req: AddReq) -> Result<AddResp, StdError>;
}
pub mod add_service {
- use std::task::Poll;
+ use std::{sync::Arc, task::Poll};
use dubbo_rust_protocol::{
jsonrpc::{Request, Response},
@@ -48,24 +48,35 @@ pub mod add_service {
use super::{AddReq, AddService, BoxFuture, StdError};
- #[derive(Clone)]
- pub struct AddServer<T: AddService + Clone> {
- inner: T,
+ type _Inner<T> = Arc<T>;
+
+ pub struct AddServer<T: AddService> {
+ inner: _Inner<T>,
+ }
+
+ impl<T: AddService> Clone for AddServer<T> {
+ fn clone(&self) -> Self {
+ Self {
+ inner: self.inner.clone(),
+ }
+ }
}
- impl<T: AddService + Clone> AddServer<T> {
+ impl<T: AddService> AddServer<T> {
pub fn new(service: T) -> Self {
- Self { inner: service }
+ Self {
+ inner: Arc::new(service),
+ }
}
}
- impl<T: AddService + Clone> NamedService for AddServer<T> {
+ impl<T: AddService> NamedService for AddServer<T> {
const SERVICE_NAME: &'static str = "AddService";
}
impl<T> tower::Service<Request> for AddServer<T>
where
- T: AddService + Clone + Send + 'static,
+ T: AddService + Send + Sync + 'static,
{
type Response = Response;
@@ -96,7 +107,7 @@ pub mod add_service {
return method_not_found(req);
}
- let mut inner_service = self.inner.clone();
+ let inner_service = self.inner.clone();
match req.header.method.as_ref().unwrap().as_str() {
"add" => Box::pin(async move {
let params: AddReq = serde_json::from_value(req.params.clone())?;
@@ -137,7 +148,7 @@ pub mod add_client {
let req_str = req.to_string()?;
let http_request = hyper::Request::builder()
.method("POST")
- .uri(format!("http://{}", self.addr.to_string()))
+ .uri(format!("http://{}/AddService", self.addr.to_string()))
.body(hyper::Body::from(req_str))
.expect("request builder");
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/client.rs b/dubbo-rust-examples/src/jsonrpc-basic/client.rs
index ef402ba..df53c27 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/client.rs
+++ b/dubbo-rust-examples/src/jsonrpc-basic/client.rs
@@ -17,7 +17,7 @@
mod addservice;
-use std::{net::SocketAddr, str::FromStr};
+use std::{net::SocketAddr, str::FromStr, time::Duration};
use addservice::{add_client::AddClient, AddReq};
@@ -27,11 +27,15 @@ async fn main() {
let mut client = AddClient::new(&addr).unwrap();
- let req = AddReq {
- numbers: vec![1, 2, 21],
- };
+ loop {
+ let req = AddReq {
+ numbers: vec![1, 2, 21],
+ };
- let resp = client.add(req).await.unwrap();
+ let resp = client.add(req).await.unwrap();
- println!("resp : {:?}", resp);
+ println!("resp : {:?}", resp);
+
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
}
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/server.rs b/dubbo-rust-examples/src/jsonrpc-basic/server.rs
index 79bd7ec..ee4d587 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/server.rs
+++ b/dubbo-rust-examples/src/jsonrpc-basic/server.rs
@@ -17,19 +17,27 @@
use dubbo_rust_protocol::jsonrpc::server::{JsonRpcServer, JsonRpcService};
use log::info;
-use std::{net::SocketAddr, str::FromStr};
+use std::{net::SocketAddr, str::FromStr, sync::Mutex};
mod addservice;
use addservice::{add_service::AddServer, AddReq, AddResp, AddService, StdError};
-
-#[derive(Clone)]
-struct MyAdd;
+struct MyAdd {
+ times: Mutex<usize>,
+}
#[async_trait::async_trait]
impl AddService for MyAdd {
- async fn add(&mut self, req: AddReq) -> Result<AddResp, StdError> {
- info!("get request {:?}", req);
+ async fn add(&self, req: AddReq) -> Result<AddResp, StdError> {
+ let times = {
+ if let Ok(mut v) = self.times.lock() {
+ *v += 1;
+ *v
+ } else {
+ 0
+ }
+ };
+ info!("get request {:?} this is no.{} call", req, times);
Ok(req.numbers.iter().sum())
}
}
@@ -44,9 +52,11 @@ async fn main() {
let addr = SocketAddr::from_str("0.0.0.0:40021").unwrap();
let rt = tokio::runtime::Handle::current();
- let clone_service = JsonRpcService::new(AddServer::new(MyAdd));
+ let service_impl = JsonRpcService::new(AddServer::new(MyAdd {
+ times: Mutex::new(0),
+ }));
- let server = JsonRpcServer::new(&addr, rt, clone_service);
+ let server = JsonRpcServer::new(&addr, rt, service_impl);
info!("Server start at {}", addr.to_string());
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs b/dubbo-rust-examples/src/jsonrpc-with-router/addservice.rs
similarity index 85%
copy from dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
copy to dubbo-rust-examples/src/jsonrpc-with-router/addservice.rs
index 91f554b..82b35b2 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
+++ b/dubbo-rust-examples/src/jsonrpc-with-router/addservice.rs
@@ -35,11 +35,11 @@ pub type AddResp = i32;
#[async_trait]
pub trait AddService {
- async fn add(&mut self, req: AddReq) -> Result<AddResp, StdError>;
+ async fn add(&self, req: AddReq) -> Result<AddResp, StdError>;
}
pub mod add_service {
- use std::task::Poll;
+ use std::{sync::Arc, task::Poll};
use dubbo_rust_protocol::{
jsonrpc::{Request, Response},
@@ -48,24 +48,35 @@ pub mod add_service {
use super::{AddReq, AddService, BoxFuture, StdError};
- #[derive(Clone)]
- pub struct AddServer<T: AddService + Clone> {
- inner: T,
+ type _Inner<T> = Arc<T>;
+
+ pub struct AddServer<T: AddService> {
+ inner: _Inner<T>,
+ }
+
+ impl<T: AddService> Clone for AddServer<T> {
+ fn clone(&self) -> Self {
+ Self {
+ inner: self.inner.clone(),
+ }
+ }
}
- impl<T: AddService + Clone> AddServer<T> {
+ impl<T: AddService> AddServer<T> {
pub fn new(service: T) -> Self {
- Self { inner: service }
+ Self {
+ inner: Arc::new(service),
+ }
}
}
- impl<T: AddService + Clone> NamedService for AddServer<T> {
+ impl<T: AddService> NamedService for AddServer<T> {
const SERVICE_NAME: &'static str = "AddService";
}
impl<T> tower::Service<Request> for AddServer<T>
where
- T: AddService + Clone + Send + 'static,
+ T: AddService + Send + Sync + 'static,
{
type Response = Response;
@@ -96,7 +107,7 @@ pub mod add_service {
return method_not_found(req);
}
- let mut inner_service = self.inner.clone();
+ let inner_service = self.inner.clone();
match req.header.method.as_ref().unwrap().as_str() {
"add" => Box::pin(async move {
let params: AddReq = serde_json::from_value(req.params.clone())?;
@@ -137,7 +148,7 @@ pub mod add_client {
let req_str = req.to_string()?;
let http_request = hyper::Request::builder()
.method("POST")
- .uri(format!("http://{}", self.addr.to_string()))
+ .uri(format!("http://{}/AddService", self.addr.to_string()))
.body(hyper::Body::from(req_str))
.expect("request builder");
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/client.rs b/dubbo-rust-examples/src/jsonrpc-with-router/client.rs
similarity index 54%
copy from dubbo-rust-examples/src/jsonrpc-basic/client.rs
copy to dubbo-rust-examples/src/jsonrpc-with-router/client.rs
index ef402ba..244a1da 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/client.rs
+++ b/dubbo-rust-examples/src/jsonrpc-with-router/client.rs
@@ -16,22 +16,33 @@
*/
mod addservice;
+mod mutilservice;
-use std::{net::SocketAddr, str::FromStr};
+use std::{net::SocketAddr, str::FromStr, time::Duration};
use addservice::{add_client::AddClient, AddReq};
+use mutilservice::{mutil_client::MutilClient, MutilReq};
#[tokio::main]
async fn main() {
- let addr = SocketAddr::from_str("127.0.0.1:40021").unwrap();
-
- let mut client = AddClient::new(&addr).unwrap();
-
- let req = AddReq {
- numbers: vec![1, 2, 21],
- };
-
- let resp = client.add(req).await.unwrap();
-
- println!("resp : {:?}", resp);
+ let addr = SocketAddr::from_str("127.0.0.1:40022").unwrap();
+
+ let mut client_add = AddClient::new(&addr).unwrap();
+ let mut client_mutil = MutilClient::new(&addr).unwrap();
+ loop {
+ let add_req = AddReq {
+ numbers: vec![1, 2, 21],
+ };
+ let resp = client_add.add(add_req).await.unwrap();
+
+ let mutil_req = MutilReq {
+ numbers: vec![1, 20, 30],
+ };
+ let resp2 = client_mutil.mutil(mutil_req).await.unwrap();
+
+ println!("add resp : {:?}", resp);
+ println!("mutil resp : {:?}", resp2);
+
+ tokio::time::sleep(Duration::from_millis(100)).await;
+ }
}
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs b/dubbo-rust-examples/src/jsonrpc-with-router/mutilservice.rs
similarity index 71%
copy from dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
copy to dubbo-rust-examples/src/jsonrpc-with-router/mutilservice.rs
index 91f554b..536a1f6 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/addservice.rs
+++ b/dubbo-rust-examples/src/jsonrpc-with-router/mutilservice.rs
@@ -27,45 +27,56 @@ pub type StdError = Box<(dyn std::error::Error + Send + Sync + 'static)>;
pub type BoxFuture<R, E> = Pin<Box<(dyn Future<Output = Result<R, E>> + Send + 'static)>>;
#[derive(Debug, Deserialize, Serialize)]
-pub struct AddReq {
+pub struct MutilReq {
pub numbers: Vec<i32>,
}
-pub type AddResp = i32;
+pub type MutilResp = i32;
#[async_trait]
-pub trait AddService {
- async fn add(&mut self, req: AddReq) -> Result<AddResp, StdError>;
+pub trait MutilService {
+ async fn mutil(&self, req: MutilReq) -> Result<MutilResp, StdError>;
}
-pub mod add_service {
- use std::task::Poll;
+pub mod mutil_service {
+ use std::{sync::Arc, task::Poll};
use dubbo_rust_protocol::{
jsonrpc::{Request, Response},
NamedService,
};
- use super::{AddReq, AddService, BoxFuture, StdError};
+ use super::{BoxFuture, MutilReq, MutilService, StdError};
- #[derive(Clone)]
- pub struct AddServer<T: AddService + Clone> {
- inner: T,
+ type _Inner<T> = Arc<T>;
+
+ pub struct MutilServer<T: MutilService> {
+ inner: _Inner<T>,
+ }
+
+ impl<T: MutilService> Clone for MutilServer<T> {
+ fn clone(&self) -> Self {
+ Self {
+ inner: self.inner.clone(),
+ }
+ }
}
- impl<T: AddService + Clone> AddServer<T> {
+ impl<T: MutilService> MutilServer<T> {
pub fn new(service: T) -> Self {
- Self { inner: service }
+ Self {
+ inner: Arc::new(service),
+ }
}
}
- impl<T: AddService + Clone> NamedService for AddServer<T> {
- const SERVICE_NAME: &'static str = "AddService";
+ impl<T: MutilService> NamedService for MutilServer<T> {
+ const SERVICE_NAME: &'static str = "MutilService";
}
- impl<T> tower::Service<Request> for AddServer<T>
+ impl<T> tower::Service<Request> for MutilServer<T>
where
- T: AddService + Clone + Send + 'static,
+ T: MutilService + Send + Sync + 'static,
{
type Response = Response;
@@ -96,11 +107,11 @@ pub mod add_service {
return method_not_found(req);
}
- let mut inner_service = self.inner.clone();
+ let inner_service = self.inner.clone();
match req.header.method.as_ref().unwrap().as_str() {
- "add" => Box::pin(async move {
- let params: AddReq = serde_json::from_value(req.params.clone())?;
- let result = inner_service.add(params).await?;
+ "mutil" => Box::pin(async move {
+ let params: MutilReq = serde_json::from_value(req.params.clone())?;
+ let result = inner_service.mutil(params).await?;
Ok(Response::from_request(&req, result)?)
}),
_ => method_not_found(req),
@@ -109,21 +120,21 @@ pub mod add_service {
}
}
-pub mod add_client {
+pub mod mutil_client {
use std::net::SocketAddr;
- use super::{AddReq, AddResp, StdError};
+ use super::{MutilReq, MutilResp, StdError};
use dubbo_rust_protocol::jsonrpc::Request as JsonRpcRequest;
use dubbo_rust_protocol::jsonrpc::Response as JsonRpcResponse;
use hyper::body::HttpBody;
use hyper::client::HttpConnector;
- pub struct AddClient {
+ pub struct MutilClient {
addr: SocketAddr,
http_client: hyper::Client<HttpConnector>,
}
- impl AddClient {
+ impl MutilClient {
pub fn new(addr: &SocketAddr) -> Result<Self, StdError> {
let client = hyper::Client::new();
Ok(Self {
@@ -132,12 +143,12 @@ pub mod add_client {
})
}
- pub async fn add(&mut self, req: AddReq) -> Result<AddResp, StdError> {
- let req = JsonRpcRequest::new("add", req)?;
+ pub async fn mutil(&mut self, req: MutilReq) -> Result<MutilResp, StdError> {
+ let req = JsonRpcRequest::new("mutil", req)?;
let req_str = req.to_string()?;
let http_request = hyper::Request::builder()
.method("POST")
- .uri(format!("http://{}", self.addr.to_string()))
+ .uri(format!("http://{}/MutilService", self.addr.to_string()))
.body(hyper::Body::from(req_str))
.expect("request builder");
diff --git a/dubbo-rust-examples/src/jsonrpc-basic/server.rs b/dubbo-rust-examples/src/jsonrpc-with-router/server.rs
similarity index 65%
copy from dubbo-rust-examples/src/jsonrpc-basic/server.rs
copy to dubbo-rust-examples/src/jsonrpc-with-router/server.rs
index 79bd7ec..f21c736 100644
--- a/dubbo-rust-examples/src/jsonrpc-basic/server.rs
+++ b/dubbo-rust-examples/src/jsonrpc-with-router/server.rs
@@ -20,20 +20,31 @@ use log::info;
use std::{net::SocketAddr, str::FromStr};
mod addservice;
+mod mutilservice;
use addservice::{add_service::AddServer, AddReq, AddResp, AddService, StdError};
+use mutilservice::{mutil_service::MutilServer, MutilReq, MutilResp, MutilService};
-#[derive(Clone)]
struct MyAdd;
#[async_trait::async_trait]
impl AddService for MyAdd {
- async fn add(&mut self, req: AddReq) -> Result<AddResp, StdError> {
+ async fn add(&self, req: AddReq) -> Result<AddResp, StdError> {
info!("get request {:?}", req);
Ok(req.numbers.iter().sum())
}
}
+struct MyMutil;
+
+#[async_trait::async_trait]
+impl MutilService for MyMutil {
+ async fn mutil(&self, req: MutilReq) -> Result<MutilResp, StdError> {
+ info!("get request {:?}", req);
+ Ok(req.numbers.iter().fold(1, |ret, v| ret * v))
+ }
+}
+
#[tokio::main]
async fn main() {
// log
@@ -41,12 +52,15 @@ async fn main() {
.filter_level(log::LevelFilter::Info)
.init();
- let addr = SocketAddr::from_str("0.0.0.0:40021").unwrap();
+ let addr = SocketAddr::from_str("0.0.0.0:40022").unwrap();
let rt = tokio::runtime::Handle::current();
- let clone_service = JsonRpcService::new(AddServer::new(MyAdd));
+ let mut builder = JsonRpcService::builder();
+ builder.add_service(AddServer::new(MyAdd)).unwrap();
+ builder.add_service(MutilServer::new(MyMutil)).unwrap();
+ let service = builder.build().unwrap();
- let server = JsonRpcServer::new(&addr, rt, clone_service);
+ let server = JsonRpcServer::new(&addr, rt, service);
info!("Server start at {}", addr.to_string());