Missing removeListener?

[svroonland]

I'm diagnosing an issue with a "dispatcher already shutdown" error message, using the 1.0.0 version of redis4cats. Here's the exception that is thrown:

Unexpected error occurred in RedisPubSubListener callback
java.lang.IllegalStateException: dispatcher already shutdown
	at cats.effect.std.Dispatcher$$anon$2.unsafeToFutureCancelable(Dispatcher.scala:283)
	at cats.effect.std.DispatcherPlatform.unsafeRunTimed(DispatcherPlatform.scala:60)
	at cats.effect.std.DispatcherPlatform.unsafeRunTimed$(DispatcherPlatform.scala:59)
	at cats.effect.std.Dispatcher$$anon$2.unsafeRunTimed(Dispatcher.scala:190)
	at cats.effect.std.DispatcherPlatform.unsafeRunSync(DispatcherPlatform.scala:52)
	at cats.effect.std.DispatcherPlatform.unsafeRunSync$(DispatcherPlatform.scala:51)
	at cats.effect.std.Dispatcher$$anon$2.unsafeRunSync(Dispatcher.scala:190)
	at dev.profunktor.redis4cats.pubsub.internals.PubSubInternals$$anon$1.message(PubSubInternals.scala:37)
	at io.lettuce.core.pubsub.PubSubEndpoint.notifyListeners(PubSubEndpoint.java:219)
	at io.lettuce.core.pubsub.PubSubEndpoint.notifyMessage(PubSubEndpoint.java:208)
	at io.lettuce.core.pubsub.PubSubCommandHandler.doNotifyMessage(PubSubCommandHandler.java:292)
	at io.lettuce.core.pubsub.PubSubCommandHandler.decode(PubSubCommandHandler.java:130)
	at io.lettuce.core.protocol.CommandHandler.channelRead(CommandHandler.java:598)
	at io.netty.channel.AbstractChannelHandlerContext.invokeChannelRead(AbstractChannelHandlerContext.java:379)

Looking at PubSubInternals, I noticed there is a addListener call but there is no matching removeListener call anywhere. Could this be the cause, and should this be treated as a Resource?

[gvolpe]

A minimal example to reproduce the issue would be great.

[svroonland]

Of course. Here is the reproducer, it's a copy of PubSubDemo with the take(1) added:

import cats.effect.IO
import dev.profunktor.redis4cats.connection._
import dev.profunktor.redis4cats.data.RedisChannel
import dev.profunktor.redis4cats.effect.Log.NoOp._
import dev.profunktor.redis4cats.pubsub.PubSub
import fs2.{ Pipe, Stream }
import scala.concurrent.duration._
import scala.util.Random

object PubSubDemo extends LoggerIOApp {

  import Demo._

  private val eventsChannel = RedisChannel("events")
  private val gamesChannel  = RedisChannel("games")

  def sink(name: String): Pipe[IO, String, Unit] =
    _.evalMap(x => putStrLn(s"Subscriber: $name >> $x"))

  val stream: Stream[IO, Unit] =
    (for {
      client <- Stream.resource(RedisClient[IO].from(redisURI))
      pubSub <- Stream.resource(PubSub.mkPubSubConnection[IO, String, String](client, stringCodec))
      sub1 = pubSub.subscribe(eventsChannel)
      sub2 = pubSub.subscribe(gamesChannel)
      pub1 = pubSub.publish(eventsChannel)
      pub2 = pubSub.publish(gamesChannel)
    } yield Stream(
      sub1.take(1).through(sink("#events")),
      sub2.through(sink("#games")),
      Stream.awakeEvery[IO](3.seconds) >> Stream.eval(IO(Random.nextInt(100).toString)).through(pub1),
      Stream.awakeEvery[IO](5.seconds) >> Stream.emit("Pac-Man!").through(pub2),
      Stream.awakeDelay[IO](11.seconds) >> pubSub.unsubscribe(gamesChannel),
      Stream.awakeEvery[IO](6.seconds) >> pubSub
            .pubSubSubscriptions(List(eventsChannel, gamesChannel))
            .evalMap(putStrLn)
    ).parJoin(6).drain).flatten

  val program: IO[Unit] =
    stream.compile.drain

}

[svroonland]

I will make an attempt at a fix and submit a PR.

[svroonland]

@gvolpe Would it be possible to make a release with this patch?

[gvolpe]

Sure, we should be cutting a release sometime soon. In the meantime, a snapshot is automatically built on every commit to master, in case you didn't know. Check the CI jobs, you can find the snapshot published there.