Merge pull request #17 from rsocket/develop

Add some examples and support Unix Socket.

Author: jjeffcaii

.gitignore

@@ -1,5 +1,5 @@
 target/
 **/*.rs.bk
 Cargo.lock
-
+.vscode/
 

Cargo.toml

@@ -5,6 +5,7 @@ members = [
 "rsocket",
 # transports
 "rsocket-transport-tcp",
+"rsocket-transport-unix",
 "rsocket-transport-websocket",
 "rsocket-transport-wasm",
 # extra

examples/Cargo.toml

@@ -10,15 +10,21 @@ log = "0.4.8"
 env_logger = "0.7.1"
 futures = "0.3.5"
 clap = "2.33.1"
+tokio-postgres = "0.5.4"
+redis = "0.16.0"
 
 [dev-dependencies.rsocket_rust]
-version = "0.5.2"
+path = "../rsocket"
 
 [dev-dependencies.rsocket_rust_transport_tcp]
-version = "0.5.2"
+path = "../rsocket-transport-tcp"
+
+[dev-dependencies.rsocket_rust_transport_unix]
+path = "../rsocket-transport-unix"
 
 [dev-dependencies.rsocket_rust_transport_websocket]
 version = "0.5.2"
+path = "../rsocket-transport-websocket"
 
 [dev-dependencies.tokio]
 version = "0.2.21"
@@ -40,3 +46,19 @@ path = "cli.rs"
 [[example]]
 name = "qps"
 path = "qps.rs"
+
+[[example]]
+name = "postgres"
+path = "postgres.rs"
+
+[[example]]
+name = "redis"
+path = "redis.rs"
+
+[[example]]
+name = "echo_uds"
+path = "echo_uds.rs"
+
+[[example]]
+name = "echo_uds_client"
+path = "echo_uds_client.rs"

examples/docker-compose.yml

@@ -0,0 +1,21 @@
+version: '3.1'
+
+services:
+  postgres:
+    image: postgres:9.6-alpine
+    environment:
+      POSTGRES_USER: postgres
+      POSTGRES_PASSWORD: postgres
+    ports:
+      - 5432:5432
+    volumes:
+      - pgdata:/var/lib/postgresql/data
+  redis:
+    image: "redis:3-alpine"
+    ports:
+      - "6379:6379"
+    command: ["redis-server", "--appendonly", "yes"]
+
+volumes:
+  pgdata:
+    driver: local

examples/echo_uds.rs

@@ -0,0 +1,28 @@
+use log::info;
+use rsocket_rust::prelude::{EchoRSocket, RSocketFactory, ServerResponder};
+use rsocket_rust_transport_unix::UnixServerTransport;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let transport: UnixServerTransport = UnixServerTransport::from("/tmp/rsocket-uds.sock");
+
+    let responder: ServerResponder = Box::new(|setup, _socket| {
+        info!("accept setup: {:?}", setup);
+        Ok(Box::new(EchoRSocket))
+        // Or you can reject setup
+        // Err(From::from("SETUP_NOT_ALLOW"))
+    });
+
+    let on_start: Box<dyn FnMut() + Send + Sync> =
+        Box::new(|| info!("+++++++ echo server started! +++++++"));
+
+    RSocketFactory::receive()
+        .transport(transport)
+        .acceptor(responder)
+        .on_start(on_start)
+        .serve()
+        .await?;
+
+    Ok(())
+}

examples/echo_uds_client.rs

@@ -0,0 +1,31 @@
+use log::info;
+use rsocket_rust::prelude::{ClientResponder, EchoRSocket, Payload, RSocket, RSocketFactory};
+use rsocket_rust_transport_unix::UnixClientTransport;
+use std::error::Error;
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let responder: ClientResponder = Box::new(|| Box::new(EchoRSocket));
+
+    let client = RSocketFactory::connect()
+        .acceptor(responder)
+        .transport(UnixClientTransport::from("/tmp/rsocket-uds.sock"))
+        .setup(Payload::from("READY!"))
+        .mime_type("text/plain", "text/plain")
+        .start()
+        .await
+        .unwrap();
+
+    let request_payload: Payload = Payload::builder()
+        .set_data_utf8("Hello World!")
+        .set_metadata_utf8("Rust")
+        .build();
+
+    let res = client.request_response(request_payload).await.unwrap();
+
+    info!("got: {:?}", res);
+
+    client.close();
+
+    Ok(())
+}

examples/postgres.rs

@@ -0,0 +1,79 @@
+#[macro_use]
+extern crate log;
+
+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().await?;
+    RSocketFactory::receive()
+        .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)]
+struct Dao {
+    client: Arc<PgClient>,
+}
+
+impl RSocket for Dao {
+    fn request_response(&self, _: Payload) -> Mono<Result<Payload, RSocketError>> {
+        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())
+        })
+    }
+
+    fn metadata_push(&self, _: Payload) -> Mono<()> {
+        unimplemented!()
+    }
+
+    fn fire_and_forget(&self, _: Payload) -> Mono<()> {
+        unimplemented!()
+    }
+
+    fn request_stream(&self, _: Payload) -> Flux<Result<Payload, RSocketError>> {
+        unimplemented!()
+    }
+
+    fn request_channel(
+        &self,
+        _: Flux<Result<Payload, RSocketError>>,
+    ) -> Flux<Result<Payload, RSocketError>> {
+        unimplemented!()
+    }
+}
+
+impl Dao {
+    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 {
+            client: Arc::new(client),
+        })
+    }
+}

examples/redis.rs

@@ -0,0 +1,64 @@
+use redis::Client as RedisClient;
+use rsocket_rust::{error::RSocketError, prelude::*};
+use rsocket_rust_transport_tcp::TcpServerTransport;
+use std::error::Error;
+use std::str::FromStr;
+
+#[derive(Clone)]
+struct RedisDao {
+    inner: RedisClient,
+}
+
+#[tokio::main]
+async fn main() -> Result<(), Box<dyn Error + Send + Sync>> {
+    let dao = RedisDao::from_str("redis://127.0.0.1").expect("Connect redis failed!");
+    RSocketFactory::receive()
+        .acceptor(Box::new(move |_setup, _socket| Ok(Box::new(dao.clone()))))
+        .transport(TcpServerTransport::from("127.0.0.1:7878"))
+        .serve()
+        .await
+}
+
+impl FromStr for RedisDao {
+    type Err = redis::RedisError;
+
+    fn from_str(s: &str) -> Result<Self, Self::Err> {
+        let client = redis::Client::open(s)?;
+        Ok(RedisDao { inner: client })
+    }
+}
+
+impl RSocket for RedisDao {
+    fn request_response(&self, req: Payload) -> Mono<Result<Payload, RSocketError>> {
+        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())
+        })
+    }
+
+    fn metadata_push(&self, _req: Payload) -> Mono<()> {
+        unimplemented!()
+    }
+    fn fire_and_forget(&self, _req: Payload) -> Mono<()> {
+        unimplemented!()
+    }
+    fn request_stream(&self, _req: Payload) -> Flux<Result<Payload, RSocketError>> {
+        unimplemented!()
+    }
+    fn request_channel(
+        &self,
+        _reqs: Flux<Result<Payload, RSocketError>>,
+    ) -> Flux<Result<Payload, RSocketError>> {
+        unimplemented!()
+    }
+}

justfile

@@ -0,0 +1,4 @@
+default:
+	echo 'Hello, world!'
+test:
+	@cargo test -- --nocapture

rsocket-benchmark/Cargo.toml

@@ -12,22 +12,22 @@ env_logger = "0.7.1"
 bytes = "0.5.4"
 hex = "0.4.2"
 rand = "0.7.3"
-serde = "1.0.110"
-serde_derive = "1.0.110"
+serde = "1.0.111"
+serde_derive = "1.0.111"
 criterion = "0.3.2"
 
 [dev-dependencies.rsocket_rust]
-version = "0.5.2"
+path = "../rsocket"
 features = ["frame"]
 
 [dev-dependencies.rsocket_rust_transport_tcp]
-version = "0.5.2"
+path = "../rsocket-transport-tcp"
 
 [dev-dependencies.rsocket_rust_transport_websocket]
-version = "0.5.2"
+path = "../rsocket-transport-websocket"
 
 [dev-dependencies.rsocket_rust_messaging]
-version = "0.5.2"
+path = "../rsocket-messaging"
 
 [dev-dependencies.tokio]
 version = "0.2.21"