feat: redesign RSocket trait based on async_trait

jjeffcaii · jjeffcaii · commit 0439e93265a1 · 2020-12-27T20:54:10.000+08:00
diff --git a/examples/Cargo.toml b/examples/Cargo.toml
@@ -11,7 +11,8 @@ env_logger = "0.8.2"
 futures = "0.3.8"
 clap = "2.33.3"
 tokio-postgres = "0.6.0"
-redis = "0.17.0"
+redis = "0.16.0"
+async-trait = "0.1.42"
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"
diff --git a/examples/cli.rs b/examples/cli.rs
@@ -2,6 +2,7 @@
 extern crate log;
 
 use rsocket_rust::prelude::*;
+use rsocket_rust::utils::EchoRSocket;
 use rsocket_rust::Result;
 use rsocket_rust_transport_tcp::TcpClientTransport;
 
diff --git a/examples/echo.rs b/examples/echo.rs
@@ -4,6 +4,7 @@ extern crate log;
 use clap::{App, Arg, SubCommand};
 use rsocket_rust::prelude::*;
 use rsocket_rust::transport::Connection;
+use rsocket_rust::utils::EchoRSocket;
 use rsocket_rust::Result;
 use rsocket_rust_transport_tcp::{
     TcpClientTransport, TcpServerTransport, UnixClientTransport, UnixServerTransport,
@@ -50,7 +51,7 @@ where
 
     match mode {
         RequestMode::FNF => {
-            cli.fire_and_forget(req).await;
+            cli.fire_and_forget(req).await?;
         }
         RequestMode::STREAM => {
             let mut results = cli.request_stream(req);
@@ -79,7 +80,7 @@ where
             }
         }
         RequestMode::REQUEST => {
-            let res = cli.request_response(req).await.expect("Request failed!");
+            let res = cli.request_response(req).await?;
             info!("{:?}", res);
         }
     }
diff --git a/examples/postgres.rs b/examples/postgres.rs
@@ -1,6 +1,7 @@
 #[macro_use]
 extern crate log;
 
+use async_trait::async_trait;
 use rsocket_rust::prelude::*;
 use rsocket_rust::Result;
 use rsocket_rust_transport_tcp::TcpServerTransport;
@@ -24,24 +25,23 @@ struct Dao {
     client: Arc<PgClient>,
 }
 
+#[async_trait]
 impl RSocket for Dao {
-    fn request_response(&self, _: Payload) -> Mono<Result<Payload>> {
+    async fn request_response(&self, _: Payload) -> Result<Payload> {
         let client = self.client.clone();
-        Box::pin(async move {
-            let row = client
-                .query_one("SELECT 'world' AS hello", &[])
-                .await
-                .expect("Execute SQL failed!");
-            let result: String = row.get("hello");
-            Ok(Payload::builder().set_data_utf8(&result).build())
-        })
+        let row = client
+            .query_one("SELECT 'world' AS hello", &[])
+            .await
+            .expect("Execute SQL failed!");
+        let result: String = row.get("hello");
+        Ok(Payload::builder().set_data_utf8(&result).build())
     }
 
-    fn metadata_push(&self, _: Payload) -> Mono<()> {
+    async fn metadata_push(&self, _: Payload) -> Result<()> {
         unimplemented!()
     }
 
-    fn fire_and_forget(&self, _: Payload) -> Mono<()> {
+    async fn fire_and_forget(&self, _: Payload) -> Result<()> {
         unimplemented!()
     }
 
diff --git a/examples/proxy.rs b/examples/proxy.rs
@@ -3,6 +3,7 @@ extern crate log;
 
 use futures::executor::block_on;
 use rsocket_rust::prelude::*;
+use rsocket_rust::utils::EchoRSocket;
 use rsocket_rust::Result;
 use rsocket_rust_transport_tcp::*;
 
diff --git a/examples/redis.rs b/examples/redis.rs
@@ -1,3 +1,4 @@
+use async_trait::async_trait;
 use redis::Client as RedisClient;
 use rsocket_rust::prelude::*;
 use rsocket_rust::Result;
@@ -28,33 +29,34 @@ impl FromStr for RedisDao {
     }
 }
 
+#[async_trait]
 impl RSocket for RedisDao {
-    fn request_response(&self, req: Payload) -> Mono<Result<Payload>> {
+    async fn request_response(&self, req: Payload) -> Result<Payload> {
         let client = self.inner.clone();
-
-        Box::pin(async move {
-            let mut conn: redis::aio::Connection = client
-                .get_async_connection()
-                .await
-                .expect("Connect redis failed!");
-            let value: String = redis::cmd("GET")
-                .arg(&[req.data_utf8()])
-                .query_async(&mut conn)
-                .await
-                .unwrap_or("<nil>".to_owned());
-            Ok(Payload::builder().set_data_utf8(&value).build())
-        })
+        let mut conn: redis::aio::Connection = client
+            .get_async_connection()
+            .await
+            .expect("Connect redis failed!");
+        let value: String = redis::cmd("GET")
+            .arg(&[req.data_utf8()])
+            .query_async(&mut conn)
+            .await
+            .unwrap_or("<nil>".to_owned());
+        Ok(Payload::builder().set_data_utf8(&value).build())
     }
 
-    fn metadata_push(&self, _req: Payload) -> Mono<()> {
+    async fn metadata_push(&self, _req: Payload) -> Result<()> {
         unimplemented!()
     }
-    fn fire_and_forget(&self, _req: Payload) -> Mono<()> {
+
+    async fn fire_and_forget(&self, _req: Payload) -> Result<()> {
         unimplemented!()
     }
+
     fn request_stream(&self, _req: Payload) -> Flux<Result<Payload>> {
         unimplemented!()
     }
+
     fn request_channel(&self, _reqs: Flux<Result<Payload>>) -> Flux<Result<Payload>> {
         unimplemented!()
     }
diff --git a/rsocket-messaging/src/requester.rs b/rsocket-messaging/src/requester.rs
@@ -259,8 +259,7 @@ impl RequestSpec {
 
     pub async fn retrieve(self) -> Result<()> {
         let (req, _mime_type, rsocket) = self.preflight()?;
-        rsocket.fire_and_forget(req).await;
-        Ok(())
+        rsocket.fire_and_forget(req).await
     }
 
     pub async fn retrieve_mono(self) -> Unpacker {
@@ -364,10 +363,16 @@ fn do_unmarshal<T>(mime_type: &MimeType, raw: &Bytes) -> Result<Option<T>>
 where
     T: Sized + DeserializeOwned,
 {
-    // TODO: support more mime types
-    match *mime_type {
-        MimeType::APPLICATION_JSON => Ok(Some(unmarshal(misc::json(), &raw.as_ref())?)),
-        MimeType::APPLICATION_CBOR => Ok(Some(unmarshal(misc::cbor(), &raw.as_ref())?)),
+    match mime_type.as_u8() {
+        Some(code) => {
+            if code == MimeType::APPLICATION_JSON.as_u8().unwrap() {
+                Ok(Some(unmarshal(misc::json(), &raw.as_ref())?))
+            } else if code == MimeType::APPLICATION_CBOR.as_u8().unwrap() {
+                Ok(Some(unmarshal(misc::cbor(), &raw.as_ref())?))
+            } else {
+                Err(RSocketError::WithDescription("unsupported mime type!".into()).into())
+            }
+        }
         _ => Err(RSocketError::WithDescription("unsupported mime type!".into()).into()),
     }
 }
@@ -376,10 +381,16 @@ fn do_marshal<T>(mime_type: &MimeType, data: &T) -> Result<Vec<u8>>
 where
     T: Sized + Serialize,
 {
-    // TODO: support more mime types
-    match *mime_type {
-        MimeType::APPLICATION_JSON => marshal(misc::json(), data),
-        MimeType::APPLICATION_CBOR => marshal(misc::cbor(), data),
+    match mime_type.as_u8() {
+        Some(code) => {
+            if code == MimeType::APPLICATION_JSON.as_u8().unwrap() {
+                marshal(misc::json(), data)
+            } else if code == MimeType::APPLICATION_CBOR.as_u8().unwrap() {
+                marshal(misc::cbor(), data)
+            } else {
+                Err(RSocketError::WithDescription("unsupported mime type!".into()).into())
+            }
+        }
         _ => Err(RSocketError::WithDescription("unsupported mime type!".into()).into()),
     }
 }
diff --git a/rsocket-test/tests/test_clients.rs b/rsocket-test/tests/test_clients.rs
@@ -3,6 +3,7 @@ extern crate log;
 
 use futures::stream;
 use rsocket_rust::prelude::*;
+use rsocket_rust::utils::EchoRSocket;
 use rsocket_rust::Client;
 use rsocket_rust_transport_tcp::{
     TcpClientTransport, TcpServerTransport, UnixClientTransport, UnixServerTransport,
@@ -216,13 +217,13 @@ async fn exec_request_response(socket: &Client) {
 async fn exec_metadata_push(socket: &Client) {
     let pa = Payload::builder().set_metadata_utf8("Hello World!").build();
     // metadata push
-    socket.metadata_push(pa).await;
+    let _ = socket.metadata_push(pa).await;
 }
 
 async fn exec_fire_and_forget(socket: &Client) {
     // request fnf
     let fnf = Payload::from("Hello World!");
-    socket.fire_and_forget(fnf).await;
+    let _ = socket.fire_and_forget(fnf).await;
 }
 
 async fn exec_request_stream(socket: &Client) {
diff --git a/rsocket-transport-wasm/src/misc.rs b/rsocket-transport-wasm/src/misc.rs
@@ -87,7 +87,7 @@ impl JsClient {
         let request: JsPayload = request.into_serde().unwrap();
 
         future_to_promise(async move {
-            inner.fire_and_forget(request.into()).await;
+            let _ = inner.fire_and_forget(request.into()).await;
             Ok(JsValue::NULL)
         })
     }
diff --git a/rsocket/Cargo.toml b/rsocket/Cargo.toml
@@ -18,6 +18,7 @@ async-trait = "0.1.42"
 dashmap = "3.11.10"
 thiserror = "1.0.22"
 anyhow = "1.0.36"
+async-stream = "0.3.0"
 
 [target.'cfg(target_arch = "wasm32")'.dependencies]
 wasm-bindgen-futures = "0.4.19"
diff --git a/rsocket/src/core/client.rs b/rsocket/src/core/client.rs
@@ -2,11 +2,12 @@ use crate::error::{RSocketError, ERR_CONN_CLOSED};
 use crate::frame::{self, Frame};
 use crate::payload::{Payload, SetupPayload, SetupPayloadBuilder};
 use crate::runtime;
-use crate::spi::{ClientResponder, Flux, Mono, RSocket};
+use crate::spi::{ClientResponder, Flux, RSocket};
 use crate::transport::{
     self, Acceptor, Connection, DuplexSocket, Reader, Splitter, Transport, Writer,
 };
 use crate::Result;
+use async_trait::async_trait;
 use futures::{future, select, FutureExt, SinkExt, StreamExt};
 use std::error::Error;
 use std::net::SocketAddr;
@@ -209,17 +210,18 @@ impl Client {
     }
 }
 
+#[async_trait]
 impl RSocket for Client {
-    fn metadata_push(&self, req: Payload) -> Mono<()> {
-        self.socket.metadata_push(req)
+    async fn metadata_push(&self, req: Payload) -> Result<()> {
+        self.socket.metadata_push(req).await
     }
 
-    fn fire_and_forget(&self, req: Payload) -> Mono<()> {
-        self.socket.fire_and_forget(req)
+    async fn fire_and_forget(&self, req: Payload) -> Result<()> {
+        self.socket.fire_and_forget(req).await
     }
 
-    fn request_response(&self, req: Payload) -> Mono<Result<Payload>> {
-        self.socket.request_response(req)
+    async fn request_response(&self, req: Payload) -> Result<Payload> {
+        self.socket.request_response(req).await
     }
 
     fn request_stream(&self, req: Payload) -> Flux<Result<Payload>> {
diff --git a/rsocket/src/core/server.rs b/rsocket/src/core/server.rs
@@ -2,10 +2,11 @@ use crate::error::RSocketError;
 use crate::frame::{self, Frame};
 use crate::payload::SetupPayload;
 use crate::runtime;
-use crate::spi::{EmptyRSocket, RSocket, ServerResponder};
+use crate::spi::{RSocket, ServerResponder};
 use crate::transport::{
     Acceptor, Connection, DuplexSocket, ServerTransport, Splitter, Transport, MIN_MTU,
 };
+use crate::utils::EmptyRSocket;
 use crate::Result;
 use std::error::Error;
 use std::future::Future;
diff --git a/rsocket/src/lib.rs b/rsocket/src/lib.rs
@@ -37,11 +37,12 @@
 //!
 //! ```no_run,ignore
 //! use rsocket_rust::prelude::*;
+//! use rsocket_rust::utils::EchoRSocket;
+//! use rsocket_rust::Result;
 //! use rsocket_rust_transport_tcp::TcpServerTransport;
-//! use std::error::Error;
 //!
 //! #[tokio::main]
-//! async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+//! async fn main() -> Result<()> {
 //!     RSocketFactory::receive()
 //!         .transport(TcpServerTransport::from("127.0.0.1:7878"))
 //!         .acceptor(Box::new(|setup, socket| {
@@ -65,11 +66,12 @@
 //!
 //! ```no_run,ignore
 //! use rsocket_rust::prelude::*;
+//! use rsocket_rust::utils::EchoRSocket;
+//! use rsocket_rust::Result;
 //! use rsocket_rust_transport_tcp::TcpClientTransport;
-//! use std::error::Error;
 //!
 //! #[tokio::main]
-//! async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+//! async fn main() -> Result<()> {
 //!     let client = RSocketFactory::connect()
 //!         .transport(TcpClientTransport::from("127.0.0.1:7878"))
 //!         .acceptor(Box::new(|| {
@@ -112,6 +114,5 @@ pub mod utils;
 
 pub type Error = Box<dyn std::error::Error + Sync + Send>;
 pub type Result<T> = anyhow::Result<T>;
-pub type PinBoxFuture<T> = std::pin::Pin<Box<dyn std::future::Future<Output = T>>>;
 
 pub use crate::core::{Client, ClientBuilder, ServerBuilder};
diff --git a/rsocket/src/spi.rs b/rsocket/src/spi.rs
diff --git a/rsocket/src/transport/socket.rs b/rsocket/src/transport/socket.rs
diff --git a/rsocket/src/utils.rs b/rsocket/src/utils.rs

Original file line number	Diff line number	Diff line change
`@@ -4,6 +4,7 @@ extern crate log;`
`4`	`4`	`use clap::{App, Arg, SubCommand};`
`5`	`5`	`use rsocket_rust::prelude::*;`
`6`	`6`	`use rsocket_rust::transport::Connection;`
	`7`	`+use rsocket_rust::utils::EchoRSocket;`
`7`	`8`	`use rsocket_rust::Result;`
`8`	`9`	`use rsocket_rust_transport_tcp::{`
`9`	`10`	`TcpClientTransport, TcpServerTransport, UnixClientTransport, UnixServerTransport,`
`@@ -50,7 +51,7 @@ where`
`50`	`51`
`51`	`52`	`match mode {`
`52`	`53`	`RequestMode::FNF => {`
`53`		`- cli.fire_and_forget(req).await;`
	`54`	`+ cli.fire_and_forget(req).await?;`
`54`	`55`	`}`
`55`	`56`	`RequestMode::STREAM => {`
`56`	`57`	`let mut results = cli.request_stream(req);`
`@@ -79,7 +80,7 @@ where`
`79`	`80`	`}`
`80`	`81`	`}`
`81`	`82`	`RequestMode::REQUEST => {`
`82`		`- let res = cli.request_response(req).await.expect("Request failed!");`
	`83`	`+ let res = cli.request_response(req).await?;`
`83`	`84`	`info!("{:?}", res);`
`84`	`85`	`}`
`85`	`86`	`}`
Original file line number	Diff line number	Diff line change
`@@ -87,7 +87,7 @@ impl JsClient {`
`87`	`87`	`let request: JsPayload = request.into_serde().unwrap();`
`88`	`88`
`89`	`89`	`future_to_promise(async move {`
`90`		`- inner.fire_and_forget(request.into()).await;`
	`90`	`+ let _ = inner.fire_and_forget(request.into()).await;`
`91`	`91`	`Ok(JsValue::NULL)`
`92`	`92`	`})`
`93`	`93`	`}`