fix Calling subscribe() multiple times on PubSubStream with the same …

…channel results in an error on close() #59

src/client/pub_sub_stream.rs

@@ -1,9 +1,5 @@
 use crate::{
-    client::{Client, ClientPreparedCommand},
-    commands::InternalPubSubCommands,
-    network::PubSubSender,
-    resp::{ByteBufSeed, CommandArgs, SingleArg, SingleArgCollection},
-    PubSubReceiver, Result,
+    client::{Client, ClientPreparedCommand}, commands::InternalPubSubCommands, network::PubSubSender, resp::{ByteBufSeed, CommandArgs, SingleArg, SingleArgCollection}, Error, PubSubReceiver, Result
 };
 use futures_util::{Stream, StreamExt};
 use serde::{
@@ -104,6 +100,12 @@ impl PubSubSplitSink {
     {
         let channels = CommandArgs::default().arg(channels).build();
 
+        for channel in &channels {
+            if self.channels.iter().any(|c| c == channel) {
+                return Err(Error::Client(format!("pub sub stream already subscribed to channel `{}`", String::from_utf8_lossy(channel))));
+            }
+        }
+
         self.client
             .subscribe_from_pub_sub_sender(&channels, &self.sender)
             .await?;
@@ -121,6 +123,12 @@ impl PubSubSplitSink {
     {
         let patterns = CommandArgs::default().arg(patterns).build();
 
+        for pattern in &patterns {
+            if self.patterns.iter().any(|p| p == pattern) {
+                return Err(Error::Client(format!("pub sub stream already subscribed to pattern `{}`", String::from_utf8_lossy(pattern))));
+            }
+        }
+
         self.client
             .psubscribe_from_pub_sub_sender(&patterns, &self.sender)
             .await?;
@@ -138,6 +146,12 @@ impl PubSubSplitSink {
     {
         let shardchannels = CommandArgs::default().arg(shardchannels).build();
 
+        for shardchannel in &shardchannels {
+            if self.shardchannels.iter().any(|c| c == shardchannel) {
+                return Err(Error::Client(format!("pub sub stream already subscribed to shard channel `{}`", String::from_utf8_lossy(shardchannel))));
+            }
+        }
+
         self.client
             .ssubscribe_from_pub_sub_sender(&shardchannels, &self.sender)
             .await?;

src/tests/pub_sub_commands.rs

@@ -768,3 +768,26 @@ async fn split() -> Result<()> {
 
     Ok(())
 }
+
+
+#[cfg_attr(feature = "tokio-runtime", tokio::test)]
+#[cfg_attr(feature = "async-std-runtime", async_std::test)]
+#[serial]
+async fn subscribe_twice() -> Result<()> {
+    let pub_sub_client = get_test_client().await?;
+    let regular_client = get_test_client().await?;
+
+    // cleanup
+    regular_client.flushdb(FlushingMode::Sync).await?;
+
+    let mut pub_sub_stream = pub_sub_client.subscribe("mychannel").await?;
+    assert!(pub_sub_stream.subscribe("mychannel").await.is_err());
+
+    pub_sub_stream.psubscribe("pattern").await?;
+    assert!(pub_sub_stream.psubscribe("pattern").await.is_err());
+
+    pub_sub_stream.ssubscribe("mychannel").await?;
+    assert!(pub_sub_stream.ssubscribe("mychannel").await.is_err());
+
+    Ok(())
+}