Merge pull request #27 from dimacurrentai/redis

Redis pub/sub.

Author: dkorolev

step06_redis/code/Cargo.toml

@@ -7,3 +7,5 @@ clap = { version = "4.0", features = ["derive"] }
 redis = { version = "0.29.2", features = ["tokio-comp"] }
 tokio = { version = "1", features = ["full"] }
 anyhow = "1.0"
+chrono = "0.4"
+futures = "0.3"

step06_redis/code/src/main.rs

@@ -1,8 +1,12 @@
+use anyhow::Result;
+use anyhow::anyhow;
+use chrono::Local;
 use clap::Parser;
+use futures::StreamExt;
 use redis::AsyncCommands;
 use std::process::ExitCode;
-use anyhow::Result;
-use anyhow::anyhow;
+use tokio::select;
+use tokio::time::{Duration, sleep};
 
 #[derive(Parser)]
 struct Args {
@@ -11,6 +15,12 @@ struct Args {
 
   #[arg(long)]
   mode: String,
+
+  #[arg(long, default_value = "5.0")]
+  subscribe_for_seconds: f64,
+
+  #[arg(long)]
+  and_publish_in_seconds: Option<f64>,
 }
 
 async fn try_check(args: &Args) -> Result<()> {
@@ -31,13 +41,72 @@ async fn try_test(args: &Args) -> Result<()> {
   (val == "value").then(|| ()).ok_or(anyhow!("Wrong value"))
 }
 
+async fn try_pub(args: &Args) -> Result<()> {
+  let client = redis::Client::open(args.redis.to_string())?;
+  let mut con = client.get_multiplexed_async_connection().await?;
+  con.publish::<_, _, ()>("redis_channel", format!("published from rust at {}", Local::now())).await?;
+  Ok(())
+}
+
+async fn try_sub(args: &Args) -> Result<()> {
+  let client = redis::Client::open(args.redis.to_string())?;
+  let mut pubsub = client.get_async_pubsub().await?;
+
+  pubsub.subscribe("redis_channel").await?;
+  let mut stream = pubsub.on_message();
+
+  let mut con = client.get_multiplexed_async_connection().await?;
+
+  let timeout = async || sleep(Duration::from_secs_f64(args.subscribe_for_seconds)).await;
+  let mut optionally_published = false;
+  let mut and_optionally_publish = async || {
+    if !optionally_published {
+      optionally_published = true;
+      if let Some(delay) = args.and_publish_in_seconds {
+        sleep(Duration::from_secs_f64(delay)).await;
+        println!("publishing from rust from a separate green thread");
+        con.publish::<_, _, ()>("redis_channel", "published from rust after a delay").await.unwrap()
+      }
+    }
+    std::future::pending::<()>().await
+  };
+
+  println!("listening to messages on `redis_channel`");
+
+  loop {
+    select! {
+      msg = stream.next() => {
+        let payload: String = msg.unwrap().get_payload()?;
+        println!(">> {}", payload);
+      },
+      _ = timeout() => {
+        println!("terminating by timeout");
+        break Ok(())
+      },
+      _ = and_optionally_publish() => {
+        unreachable!()
+      }
+    }
+  }
+}
+
 #[tokio::main]
 async fn main() -> ExitCode {
   let args = Args::parse();
 
   match args.mode.as_str() {
     "check" => try_check(&args).await,
     "test" => try_test(&args).await,
-    _ => Err(anyhow!("This `--mode` value is not supported."))
-  }.map_or_else(|err| { println!("Error: {}", err); 1 }, |_| 0).into()
+    "pub" => try_pub(&args).await,
+    "sub" => try_sub(&args).await,
+    _ => Err(anyhow!("This `--mode` value is not supported.")),
+  }
+  .map_or_else(
+    |err| {
+      println!("Error: {}", err);
+      1
+    },
+    |_| 0,
+  )
+  .into()
 }

step06_redis/run.sh

@@ -27,5 +27,9 @@ echo 'Running the test.'
 docker run --add-host=host.docker.internal:host-gateway --rm -t demo --mode test --redis redis://host.docker.internal
 echo 'Test run successfully.'
 
+echo 'Running the pubsub test.'
+docker run --add-host=host.docker.internal:host-gateway --rm -t demo --mode sub --and-publish-in-seconds=2.5 --redis redis://host.docker.internal
+echo 'The pubsub test run successfully.'
+
 docker stop redis-for-rust
 echo 'Redis stopped.'