rename OutputStream to SubscriptionClient, consume SimpleClient by SubscriptionClient and back, rename SimpleClient::stream to SimpleClient::subscribe

Author: estin

examples/pubsub.rs

@@ -1,25 +1,20 @@
 use ntex_redis::{cmd, RedisConnector};
 use std::error::Error;
-use std::rc::Rc;
 
 #[ntex::main]
 async fn main() -> Result<(), Box<dyn Error>> {
     env_logger::init();
 
     // subscriber
-    let client = Rc::new(
-        RedisConnector::new("127.0.0.1:6379")
-            .connect_simple()
-            .await?,
-    );
+    let client = RedisConnector::new("127.0.0.1:6379")
+        .connect_simple()
+        .await?;
 
-    let client_clone = client.clone();
+    let pubsub = client.subscribe(cmd::Subscribe("pubsub")).unwrap();
 
     ntex::rt::spawn(async move {
-        let subscriber = client_clone.stream(cmd::Subscribe("pubsub")).unwrap();
-
         loop {
-            match subscriber.recv().await {
+            match pubsub.recv().await {
                 Some(Ok(cmd::SubscribeItem::Subscribed(channel))) => {
                     println!("sub: subscribed to {:?}", channel)
                 }
@@ -47,11 +42,5 @@ async fn main() -> Result<(), Box<dyn Error>> {
         redis.exec(cmd::Publish("pubsub", &value)).await?;
     }
 
-    // unsubscribe
-    client.send(cmd::UnSubscribe("pubsub"))?;
-
-    // allow to subscriber recv unsubscribe message
-    ntex::time::sleep(ntex::time::Millis(10)).await;
-
     Ok(())
 }

src/cmd/mod.rs

@@ -18,7 +18,9 @@ pub use self::connection::{Ping, Select};
 pub use self::hashes::{HDel, HGet, HGetAll, HIncrBy, HLen, HSet};
 pub use self::keys::{Del, Exists, Expire, ExpireAt, Keys, Ttl, TtlResult};
 pub use self::lists::{LIndex, LPop, LPush, RPop, RPush};
-pub use self::pubsub::{Publish, Subscribe, SubscribeItem, UnSubscribe};
+pub use self::pubsub::{
+    PubSubCommand, Publish, Subscribe, SubscribeItem, SubscribeOutputCommand, UnSubscribe,
+};
 pub use self::strings::{Get, IncrBy, Set};
 
 /// Trait implemented by types that can be used as redis commands

src/cmd/pubsub.rs

@@ -8,6 +8,8 @@ const TYPE_SUBSCRIBE: Bytes = Bytes::from_static(b"subscribe");
 const TYPE_UNSUBSCRIBE: Bytes = Bytes::from_static(b"unsubscribe");
 const TYPE_MESSAGE: Bytes = Bytes::from_static(b"message");
 
+pub trait PubSubCommand {}
+
 #[derive(Debug, PartialEq)]
 pub enum SubscribeItem {
     Subscribed(Bytes),
@@ -77,6 +79,20 @@ impl Command for SubscribeOutputCommand {
     }
 }
 
+pub struct UnSubscribeOutputCommand(pub(crate) Request);
+
+impl Command for UnSubscribeOutputCommand {
+    type Output = SubscribeItem;
+
+    fn to_request(self) -> Request {
+        self.0
+    }
+
+    fn to_output(val: Response) -> Result<Self::Output, CommandError> {
+        SubscribeItem::try_from(val)
+    }
+}
+
 /// Publish redis command
 pub fn Publish<T, V>(key: T, value: V) -> utils::IntOutputCommand
 where
@@ -101,12 +117,15 @@ where
 }
 
 /// Unsubscribe redis command
-pub fn UnSubscribe<T>(key: T) -> SubscribeOutputCommand
+pub fn UnSubscribe<T>(key: T) -> UnSubscribeOutputCommand
 where
     BulkString: From<T>,
 {
-    SubscribeOutputCommand(Request::Array(vec![
+    UnSubscribeOutputCommand(Request::Array(vec![
         Request::from_static("UNSUBSCRIBE"),
         Request::BulkString(key.into()),
     ]))
 }
+
+impl PubSubCommand for SubscribeOutputCommand {}
+impl PubSubCommand for UnSubscribeOutputCommand {}

src/lib.rs

@@ -40,7 +40,7 @@ mod simple;
 
 pub use self::client::{Client, CommandResult};
 pub use self::connector::RedisConnector;
-pub use self::simple::SimpleClient;
+pub use self::simple::{SimpleClient, SubscriptionClient};
 
 /// Macro to create a request array, useful for preparing commands to send. Elements can be any type, or a mixture
 /// of types, that satisfy `Into<Request>`.

src/simple.rs

@@ -1,7 +1,7 @@
 use std::pin::Pin;
 use std::task::{Context, Poll};
 
-use super::cmd::Command;
+use super::cmd::{Command, PubSubCommand, SubscribeOutputCommand};
 use super::codec::Codec;
 use super::errors::{CommandError, Error};
 use ntex::{io::IoBoxed, io::RecvError, util::poll_fn, util::ready, util::Stream};
@@ -35,13 +35,13 @@ impl SimpleClient {
         Ok(())
     }
 
-    /// Execute redis command and act with output as stream
-    pub fn stream<U>(&self, cmd: U) -> Result<OutputStream<U>, CommandError>
-    where
-        U: Command,
-    {
+    /// Execute redis SUBSCRIBE command and act with output as stream
+    pub fn subscribe(
+        self,
+        cmd: SubscribeOutputCommand,
+    ) -> Result<SubscriptionClient<SubscribeOutputCommand>, CommandError> {
         self.send(cmd)?;
-        Ok(OutputStream {
+        Ok(SubscriptionClient {
             client: self,
             _cmd: std::marker::PhantomData,
         })
@@ -84,20 +84,31 @@ impl SimpleClient {
     }
 }
 
-pub struct OutputStream<'a, U> {
-    client: &'a SimpleClient,
+/// Redis pubsub client to receive push messages
+pub struct SubscriptionClient<U: Command + PubSubCommand> {
+    client: SimpleClient,
     _cmd: std::marker::PhantomData<U>,
 }
 
-impl<'a, U: Command> Stream for OutputStream<'a, U> {
+impl<U: Command + PubSubCommand> Stream for SubscriptionClient<U> {
     type Item = Result<U::Output, CommandError>;
 
     fn poll_next(self: Pin<&mut Self>, cx: &mut Context<'_>) -> Poll<Option<Self::Item>> {
         self.poll_recv(cx)
     }
 }
 
-impl<'a, U: Command> OutputStream<'a, U> {
+impl<U: Command + PubSubCommand> SubscriptionClient<U> {
+    /// Get client back. Be sure to all pubsub messages from redis are received!
+    pub fn into_client(self) -> SimpleClient {
+        self.client
+    }
+
+    /// Send redis subscribe/unsubscribe command
+    pub fn send<T: Command + PubSubCommand>(&self, cmd: T) -> Result<(), CommandError> {
+        self.client.send(cmd)
+    }
+
     /// Attempt to pull out the next value of this stream.
     pub async fn recv(&self) -> Option<Result<U::Output, CommandError>> {
         poll_fn(|cx| self.client.poll_recv::<U>(cx)).await

tests/test_redis.rs

@@ -211,9 +211,9 @@ async fn test_pubsub() {
         .await
         .unwrap();
 
-    let stream = subscriber.stream(cmd::Subscribe(&channel)).unwrap();
-
-    let message = stream.recv().await;
+    // sub
+    let pubsub = subscriber.subscribe(cmd::Subscribe(&channel)).unwrap();
+    let message = pubsub.recv().await;
     assert_eq!(
         message.unwrap().unwrap(),
         cmd::SubscribeItem::Subscribed(channel.clone())
@@ -225,8 +225,8 @@ async fn test_pubsub() {
     let result = publisher.exec(cmd::Publish(&channel, "1")).await.unwrap();
     assert_eq!(result, 1);
 
-    // sub
-    let message = stream.recv().await;
+    // receive message
+    let message = pubsub.recv().await;
     assert_eq!(
         message.unwrap().unwrap(),
         cmd::SubscribeItem::Message {
@@ -236,9 +236,8 @@ async fn test_pubsub() {
     );
 
     // unsub
-    subscriber.send(cmd::UnSubscribe(&channel)).unwrap();
-
-    let message = stream.recv().await;
+    pubsub.send(cmd::UnSubscribe(&channel)).unwrap();
+    let message = pubsub.recv().await;
     assert_eq!(
         message.unwrap().unwrap(),
         cmd::SubscribeItem::UnSubscribed(channel.clone())