remove Sync trait for Mono and Flux.

Author: jjeffcaii

examples/Cargo.toml

@@ -10,8 +10,7 @@ log = "0.4.8"
 env_logger = "0.7.1"
 futures = "0.3.5"
 clap = "2.33.1"
-postgres = "0.17.3"
-r2d2_postgres = "0.16.0"
+tokio-postgres = "0.5.4"
 
 [dev-dependencies.rsocket_rust]
 path = "../rsocket"

examples/postgres.rs

@@ -1,43 +1,36 @@
 #[macro_use]
 extern crate log;
 
-use postgres::NoTls;
-use r2d2_postgres::{
-    r2d2::{self, Pool},
-    PostgresConnectionManager,
-};
 use rsocket_rust::{error::RSocketError, prelude::*};
 use rsocket_rust_transport_tcp::TcpServerTransport;
 use std::error::Error;
+use std::sync::Arc;
+use tokio_postgres::{Client as PgClient, NoTls};
 
 #[tokio::main]
 async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
     env_logger::builder().format_timestamp_millis().init();
-    let dao = Dao::try_new().expect("Connect failed!");
+    let dao = Dao::try_new().await?;
     RSocketFactory::receive()
-        .acceptor(Box::new(move |_, _| {
-            info!("accept new socket!");
-            Ok(Box::new(dao.clone()))
-        }))
+        .acceptor(Box::new(move |_, _| Ok(Box::new(dao.clone()))))
         .on_start(Box::new(|| info!("server start success!!!")))
         .transport(TcpServerTransport::from("127.0.0.1:7878"))
         .serve()
         .await
 }
 
-#[derive(Clone, Debug)]
+#[derive(Clone)]
 struct Dao {
-    pool: Pool<PostgresConnectionManager<NoTls>>,
+    client: Arc<PgClient>,
 }
 
 impl RSocket for Dao {
     fn request_response(&self, _: Payload) -> Mono<Result<Payload, RSocketError>> {
-        let pool = self.pool.clone();
+        let client = self.client.clone();
         Box::pin(async move {
-            // TODO: something wrong here!!!
-            let mut client = pool.get().expect("Get client from pool failed!");
             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())
@@ -65,16 +58,22 @@ impl RSocket for Dao {
 }
 
 impl Dao {
-    fn try_new() -> Result<Dao, Box<dyn Error + Sync + Send>> {
-        let manager: PostgresConnectionManager<NoTls> = PostgresConnectionManager::new(
-            "host=localhost user=postgres password=postgres"
-                .parse()
-                .unwrap(),
-            NoTls,
-        );
-        let pool: Pool<PostgresConnectionManager<NoTls>> =
-            r2d2::Pool::new(manager).expect("Create pool failed!");
+    async fn try_new() -> Result<Dao, Box<dyn Error + Sync + Send>> {
+        let (client, connection) =
+            tokio_postgres::connect("host=localhost user=postgres password=postgres", NoTls)
+                .await?;
+
+        // The connection object performs the actual communication with the database,
+        // so spawn it off to run on its own.
+        tokio::spawn(async move {
+            if let Err(e) = connection.await {
+                eprintln!("connection error: {}", e);
+            }
+        });
+
         info!("==> create postgres pool success!");
-        Ok(Dao { pool })
+        Ok(Dao {
+            client: Arc::new(client),
+        })
     }
 }

rsocket/src/spi.rs

@@ -18,8 +18,8 @@ pub type ServerResponder = Box<
         + Fn(SetupPayload, Box<dyn RSocket>) -> Result<Box<dyn RSocket>, Box<dyn Error>>,
 >;
 
-pub type Mono<T> = Pin<Box<dyn Send + Sync + Future<Output = T>>>;
-pub type Flux<T> = Pin<Box<dyn Send + Sync + Stream<Item = T>>>;
+pub type Mono<T> = Pin<Box<dyn Send + Future<Output = T>>>;
+pub type Flux<T> = Pin<Box<dyn Send + Stream<Item = T>>>;
 
 pub trait RSocket: Sync + Send {
     fn metadata_push(&self, req: Payload) -> Mono<()>;