Bug/subscribing to channel and pattern doubles messages in channel subscriber

.gitignore

@@ -37,6 +37,7 @@ TAGS
 lib_managed
 logs
 project/boot/*
+project/**/metals.sbt
 project/plugins/project
 src_managed
 target

modules/streams/src/main/scala/dev/profunktor/redis4cats/pubsub/internals/PubSubInternals.scala

@@ -46,7 +46,9 @@ object PubSubInternals {
             case _: IllegalStateException => throw DispatcherAlreadyShutdown()
           }
         }
-      override def message(pattern: K, channel: K, message: V): Unit = this.message(channel, message)
+
+      // Do not uncomment this, as if you will do this the channel listener will get a message twice
+      // override def message(pattern: K, channel: K, message: V): Unit = {}
     }
   private[redis4cats] def patternListener[F[_]: Async, K, V](
       redisPattern: RedisPattern[K],

modules/tests/src/test/scala/dev/profunktor/redis4cats/Redis4CatsFunSuite.scala

@@ -26,6 +26,8 @@ import dev.profunktor.redis4cats.streams.{ RedisStream, Streaming }
 
 import scala.concurrent.duration.Duration
 import scala.concurrent.{ Await, Future }
+import dev.profunktor.redis4cats.pubsub.{PubSub, PubSubCommands}
+import dev.profunktor.redis4cats.Redis4CatsFunSuite.Fs2PubSub
 
 abstract class Redis4CatsFunSuite(isCluster: Boolean) extends IOSuite {
 
@@ -51,13 +53,20 @@ abstract class Redis4CatsFunSuite(isCluster: Boolean) extends IOSuite {
   def withRedisClient[A](f: RedisClient => IO[A]): Future[Unit] =
     RedisClient[IO].from("redis://localhost").use(f).as(assert(true)).unsafeToFuture()
 
-  def withRedisStream[A](f: (Fs2Streaming[String, String], Fs2Streaming[String, String]) => IO[A]): Future[Unit] =
+  def withRedisPubSub(f: Fs2PubSub[String, String] => IO[Unit]): Future[Unit] =
+    (for {
+      client <- fs2.Stream.resource(RedisClient[IO].from("redis://localhost"))
+      pubSub <- fs2.Stream.resource(PubSub.mkPubSubConnection[IO, String, String](client, stringCodec))
+      _ <- fs2.Stream.eval(f(pubSub))
+    } yield ()).compile.drain.void.unsafeToFuture()
+
+  def withRedisStream(f: (Fs2Streaming[String, String], Fs2Streaming[String, String]) => IO[Unit]): Future[Unit] =
     (for {
       client <- fs2.Stream.resource(RedisClient[IO].from("redis://localhost"))
       readStream <- RedisStream.mkStreamingConnection[IO, String, String](client, stringCodec)
       writeStream <- RedisStream.mkStreamingConnection[IO, String, String](client, stringCodec)
-      results <- fs2.Stream.eval(f(readStream, writeStream))
-    } yield results).compile.drain.void.unsafeToFuture()
+      _ <- fs2.Stream.eval(f(readStream, writeStream))
+    } yield ()).compile.drain.void.unsafeToFuture()
 
   private def flushAll(): Future[Unit] =
     if (isCluster) withRedisCluster(_.flushAll)
@@ -88,6 +97,7 @@ abstract class Redis4CatsFunSuite(isCluster: Boolean) extends IOSuite {
 
 }
 object Redis4CatsFunSuite {
+  type Fs2PubSub[K, V] = PubSubCommands[fs2.Stream[IO, *], K, V]
 
   type Fs2Streaming[K, V] = Streaming[fs2.Stream[IO, *], K, V]
 }

modules/tests/src/test/scala/dev/profunktor/redis4cats/RedisPubSubSpec.scala

@@ -0,0 +1,58 @@
+/*
+ * Copyright 2018-2021 ProfunKtor
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+package dev.profunktor.redis4cats
+
+import dev.profunktor.redis4cats.data.{RedisChannel, RedisPattern, RedisPatternEvent}
+import cats.effect.IO
+import cats.effect.kernel.Deferred
+import scala.concurrent.duration._
+
+class RedisPubSubSpec extends Redis4CatsFunSuite(isCluster = false) {
+  test("subscribing to channel and pattern should not double messages") {
+    // Test for a bug where subscribing to a channel and pattern matching the same channel would send a message twice
+    // to the channel listener.
+
+    val channel = RedisChannel("pubsub-spec-channel-and-pattern:1")
+    val pattern = RedisPattern("pubsub-spec-channel-and-pattern:*")
+
+    case class Results(
+      channel: Vector[String],
+      pattern: Vector[RedisPatternEvent[String, String]]
+    )
+
+    withRedisPubSub { pubSub =>
+      val actualIO = for {
+        finished <- Deferred[IO, Either[Throwable, Unit]]
+        s1 <- pubSub.subscribe(channel).interruptWhen(finished).compile.to(Vector).start
+        s2 <- pubSub.psubscribe(pattern).interruptWhen(finished).compile.to(Vector).start
+        _ <- IO.sleep(200.millis) // wait for the subscription to start
+        _ <- fs2.Stream.emit("hello").through(pubSub.publish(channel)).compile.drain
+        _ <- IO.sleep(200.millis) // wait for the message to arrive
+        _ <- finished.complete(Right(()))
+        channelResults <- s1.joinWith(IO.raiseError(new RuntimeException("s1 should not be cancelled")))
+        patternResults <- s2.joinWith(IO.raiseError(new RuntimeException("s2 should not be cancelled")))
+      } yield Results(channelResults, patternResults)
+
+      val expected = Results(
+        channel = Vector("hello"), 
+        pattern = Vector(RedisPatternEvent(pattern.underlying, channel.underlying, "hello"))
+      )
+
+      actualIO.map(assertEquals(_, expected))
+    }
+  }
+}

modules/tests/src/test/scala/dev/profunktor/redis4cats/RedisStreamSpec.scala

@@ -38,7 +38,7 @@ class RedisStreamSpec extends Redis4CatsFunSuite(false) {
   private def readWriteTest(streamKey: String, length: Long): IO[Unit] =
     IO.fromFuture {
       IO {
-        withRedisStream[Unit] { (readStream, writeStream) =>
+        withRedisStream { (readStream, writeStream) =>
           val read = readStream.read(Set(streamKey), 1)
           val write =
             writeStream.append(fs2.Stream(XAddMessage(streamKey, Map("hello" -> "world"))).repeatN(length))

modules/tests/src/test/scala/dev/profunktor/redis4cats/TestScenarios.scala

@@ -280,7 +280,7 @@ trait TestScenarios { self: FunSuite =>
       _ <- IO(assert(e.nonEmpty))
       _ <- IO.sleep(50.millis)
       f <- redis.ttl("f2")
-      _ <- IO(assert(f.isEmpty))
+      _ <- IO(assertEquals(f, None))
       _ <- redis.set("f3", "yay")
       expiref3 <- redis.expire("f3", 50.millis)
       _ <- IO(assertEquals(expiref3, true))