Implement PubSub api

benchmarks/src/main/scala/zio/redis/benchmarks/BenchmarkRuntime.scala

@@ -35,6 +35,7 @@ object BenchmarkRuntime {
   private final val Layer =
     ZLayer.make[Redis](
       RedisExecutor.local,
+      RedisPubSub.local,
       ZLayer.succeed[BinaryCodec](ProtobufCodec),
       RedisLive.layer
     )

example/src/main/scala/example/Main.scala

@@ -21,7 +21,7 @@ import example.config.AppConfig
 import sttp.client3.httpclient.zio.HttpClientZioBackend
 import zhttp.service.Server
 import zio._
-import zio.redis.{RedisExecutor, RedisLive}
+import zio.redis.{RedisExecutor, RedisLive, RedisPubSub}
 import zio.schema.codec.{BinaryCodec, ProtobufCodec}
 
 object Main extends ZIOAppDefault {

redis/src/main/scala/zio/redis/ClusterExecutor.scala

@@ -150,9 +150,11 @@ object ClusterExecutor {
     } yield ExecutorScope(executor, closableScope)
 
   private def redis(address: RedisUri) = {
-    val executorLayer = ZLayer.succeed(RedisConfig(address.host, address.port)) >>> RedisExecutor.layer
-    val codecLayer    = ZLayer.succeed[BinaryCodec](StringUtf8Codec)
-    val redisLayer    = executorLayer ++ codecLayer >>> RedisLive.layer
+    val redisConfigLayer = ZLayer.succeed(RedisConfig(address.host, address.port))
+    val codecLayer       = ZLayer.succeed[BinaryCodec](StringUtf8Codec)
+    val executorLayer    = redisConfigLayer >>> RedisExecutor.layer
+    val pubSubLayer      = redisConfigLayer ++ codecLayer >>> RedisPubSub.layer
+    val redisLayer       = executorLayer ++ pubSubLayer ++ codecLayer >>> RedisLive.layer
     for {
       closableScope <- Scope.make
       layer         <- closableScope.extend[Any](redisLayer.memoize)

redis/src/main/scala/zio/redis/Output.scala

@@ -827,4 +827,50 @@ object Output {
         case other => throw ProtocolError(s"$other isn't an array")
       }
   }
+
+  case object PushProtocolOutput extends Output[PushProtocol] {
+    protected def tryDecode(respValue: RespValue)(implicit codec: BinaryCodec): PushProtocol =
+      respValue match {
+        case RespValue.NullArray => throw ProtocolError(s"Array must not be empty")
+        case RespValue.Array(values) =>
+          val name = MultiStringOutput.unsafeDecode(values(0))
+          val key  = MultiStringOutput.unsafeDecode(values(1))
+          name match {
+            case "subscribe" =>
+              val num = LongOutput.unsafeDecode(values(2))
+              PushProtocol.Subscribe(key, num)
+            case "psubscribe" =>
+              val num = LongOutput.unsafeDecode(values(2))
+              PushProtocol.PSubscribe(key, num)
+            case "unsubscribe" =>
+              val num = LongOutput.unsafeDecode(values(2))
+              PushProtocol.Unsubscribe(key, num)
+            case "punsubscribe" =>
+              val num = LongOutput.unsafeDecode(values(2))
+              PushProtocol.PUnsubscribe(key, num)
+            case "message" =>
+              val message = values(2)
+              PushProtocol.Message(key, message)
+            case "pmessage" =>
+              val channel = MultiStringOutput.unsafeDecode(values(2))
+              val message = values(3)
+              PushProtocol.PMessage(key, channel, message)
+            case other => throw ProtocolError(s"$other isn't a pushed message")
+          }
+        case other => throw ProtocolError(s"$other isn't an array")
+      }
+  }
+
+  case object NumSubResponseOutput extends Output[Chunk[NumSubResponse]] {
+    protected def tryDecode(respValue: RespValue)(implicit codec: BinaryCodec): Chunk[NumSubResponse] =
+      respValue match {
+        case RespValue.Array(values) =>
+          Chunk.fromIterator(values.grouped(2).map { chunk =>
+            val channel           = MultiStringOutput.unsafeDecode(chunk(0))
+            val numOfSubscription = LongOutput.unsafeDecode(chunk(1))
+            NumSubResponse(channel, numOfSubscription)
+          })
+        case other => throw ProtocolError(s"$other isn't an array")
+      }
+  }
 }

redis/src/main/scala/zio/redis/PubSubCommand.scala

@@ -0,0 +1,29 @@
+package zio.redis
+
+import zio.schema.codec.BinaryCodec
+import zio.stream._
+import zio.{IO, ZLayer}
+
+final case class RedisPubSubCommand(command: PubSubCommand, codec: BinaryCodec, executor: RedisPubSub) {
+  def run: IO[RedisError, List[Stream[RedisError, PushProtocol]]] = {
+    val codecLayer = ZLayer.succeed(codec)
+    executor.execute(command).provideLayer(codecLayer)
+  }
+}
+
+sealed trait PubSubCommand
+
+object PubSubCommand {
+  case class Subscribe(
+    channel: String,
+    channels: List[String],
+    onSubscribe: PubSubCallback
+  ) extends PubSubCommand
+  case class PSubscribe(
+    pattern: String,
+    patterns: List[String],
+    onSubscribe: PubSubCallback
+  ) extends PubSubCommand
+  case class Unsubscribe(channels: List[String])  extends PubSubCommand
+  case class PUnsubscribe(patterns: List[String]) extends PubSubCommand
+}

redis/src/main/scala/zio/redis/Redis.scala

@@ -31,14 +31,22 @@ trait Redis
     with api.SortedSets
     with api.Streams
     with api.Scripting
-    with api.Cluster {
+    with api.Cluster
+    with api.PubSub {
   def codec: BinaryCodec
   def executor: RedisExecutor
+  def pubSub: RedisPubSub
 }
 
-final case class RedisLive(codec: BinaryCodec, executor: RedisExecutor) extends Redis
+final case class RedisLive(codec: BinaryCodec, executor: RedisExecutor, pubSub: RedisPubSub) extends Redis
 
 object RedisLive {
-  lazy val layer: URLayer[RedisExecutor with BinaryCodec, Redis] =
-    ZLayer.fromFunction(RedisLive.apply _)
+  lazy val layer: URLayer[RedisPubSub with RedisExecutor with BinaryCodec, Redis] =
+    ZLayer.fromZIO(
+      for {
+        codec    <- ZIO.service[BinaryCodec]
+        executor <- ZIO.service[RedisExecutor]
+        pubSub   <- ZIO.service[RedisPubSub]
+      } yield RedisLive(codec, executor, pubSub)
+    )
 }

redis/src/main/scala/zio/redis/RedisEnvironment.scala

@@ -5,4 +5,5 @@ import zio.schema.codec.BinaryCodec
 private[redis] trait RedisEnvironment {
   def codec: BinaryCodec
   def executor: RedisExecutor
+  def pubSub: RedisPubSub
 }

redis/src/main/scala/zio/redis/RedisError.scala

@@ -46,5 +46,7 @@ object RedisError {
   object Moved {
     def apply(slotAndAddress: (Slot, RedisUri)): Moved = Moved(slotAndAddress._1, slotAndAddress._2)
   }
-  final case class IOError(exception: IOException) extends RedisError
+  final case class NoPubSubStream(key: String)       extends RedisError
+  final case class NoUnsubscribeRequest(key: String) extends RedisError
+  final case class IOError(exception: IOException)   extends RedisError
 }

redis/src/main/scala/zio/redis/RedisExecutor.scala

@@ -8,8 +8,8 @@ trait RedisExecutor {
 
 object RedisExecutor {
   lazy val layer: ZLayer[RedisConfig, RedisError.IOError, RedisExecutor] =
-    RedisConnectionLive.layer >>> SingleNodeExecutor.layer
+    RedisConnectionLive.layer.fresh >>> SingleNodeExecutor.layer
 
   lazy val local: ZLayer[Any, RedisError.IOError, RedisExecutor] =
-    RedisConnectionLive.default >>> SingleNodeExecutor.layer
+    RedisConnectionLive.default.fresh >>> SingleNodeExecutor.layer
 }

redis/src/main/scala/zio/redis/RedisPubSub.scala

@@ -0,0 +1,22 @@
+package zio.redis
+
+import zio.schema.codec.BinaryCodec
+import zio.stream._
+import zio.{ZIO, ZLayer}
+
+trait RedisPubSub {
+  def execute(command: PubSubCommand): ZIO[BinaryCodec, RedisError, List[Stream[RedisError, PushProtocol]]]
+}
+
+object RedisPubSub {
+  lazy val layer: ZLayer[RedisConfig with BinaryCodec, RedisError.IOError, RedisPubSub] =
+    RedisConnectionLive.layer.fresh >>> pubSublayer
+
+  lazy val local: ZLayer[BinaryCodec, RedisError.IOError, RedisPubSub] =
+    RedisConnectionLive.default.fresh >>> pubSublayer
+
+  private lazy val pubSublayer: ZLayer[RedisConnection with BinaryCodec, RedisError.IOError, RedisPubSub] =
+    ZLayer.scoped(
+      ZIO.service[RedisConnection].flatMap(SingleNodeRedisPubSub.create(_))
+    )
+}

redis/src/main/scala/zio/redis/ResultBuilder.scala

@@ -19,6 +19,7 @@ package zio.redis
 import zio.IO
 import zio.redis.ResultBuilder.NeedsReturnType
 import zio.schema.Schema
+import zio.stream.Stream
 
 sealed trait ResultBuilder {
   final def map(f: Nothing => Any)(implicit nrt: NeedsReturnType): IO[Nothing, Nothing] = ???
@@ -46,4 +47,8 @@ object ResultBuilder {
   trait ResultOutputBuilder extends ResultBuilder {
     def returning[R: Output]: IO[RedisError, R]
   }
+
+  trait ResultStreamBuilder[+F[_]] {
+    def returning[R: Schema]: IO[RedisError, F[Stream[RedisError, R]]]
+  }
 }