add redis example.

Author: jjeffcaii

examples/Cargo.toml

@@ -11,6 +11,7 @@ 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]
 path = "../rsocket"
@@ -46,3 +47,7 @@ path = "qps.rs"
 [[example]]
 name = "postgres"
 path = "postgres.rs"
+
+[[example]]
+name = "redis"
+path = "redis.rs"

examples/docker-compose.yml

@@ -10,8 +10,12 @@ services:
       - 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/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!()
+    }
+}

rsocket/Cargo.toml

@@ -14,6 +14,7 @@ log = "0.4.8"
 bytes = "0.5.4"
 futures = "0.3.5"
 lazy_static = "1.4.0"
+async-trait = "0.1.33"
 
 [dependencies.tokio]
 version = "0.2.21"