Reorganize internals and layers usage (zio#788)

Author: mijicd

README.md

@@ -4,7 +4,7 @@
 
 # ZIO Redis
 
-[![Development](https://img.shields.io/badge/Project%20Stage-Development-green.svg)](https://github.com/zio/zio/wiki/Project-Stages) ![CI Badge](https://github.com/zio/zio-redis/workflows/CI/badge.svg) [![Sonatype Releases](https://img.shields.io/nexus/r/https/oss.sonatype.org/dev.zio/zio-redis_2.13.svg?label=Sonatype%20Release)](https://oss.sonatype.org/content/repositories/releases/dev/zio/zio-redis_2.13/) [![Sonatype Snapshots](https://img.shields.io/nexus/s/https/oss.sonatype.org/dev.zio/zio-redis_2.13.svg?label=Sonatype%20Snapshot)](https://oss.sonatype.org/content/repositories/snapshots/dev/zio/zio-redis_2.13/) [![javadoc](https://javadoc.io/badge2/dev.zio/zio-redis-docs_2.13/javadoc.svg)](https://javadoc.io/doc/dev.zio/zio-redis-docs_2.13) [![ZIO Redis](https://img.shields.io/github/stars/zio/zio-redis?style=social)](https://github.com/zio/zio-redis)
+[![Development](https://img.shields.io/badge/Project%20Stage-Development-green.svg)](https://github.com/zio/zio/wiki/Project-Stages) ![CI Badge](https://github.com/zio/zio-redis/workflows/CI/badge.svg) [![Sonatype Snapshots](https://img.shields.io/nexus/s/https/oss.sonatype.org/dev.zio/zio-redis_2.13.svg?label=Sonatype%20Snapshot)](https://oss.sonatype.org/content/repositories/snapshots/dev/zio/zio-redis_2.13/) [![ZIO Redis](https://img.shields.io/github/stars/zio/zio-redis?style=social)](https://github.com/zio/zio-redis)
 
 ## Introduction
 
@@ -17,7 +17,7 @@ instances.
 To use ZIO Redis, add the following line to your `build.sbt`:
 
 ```scala
-libraryDependencies += "dev.zio" %% "zio-redis" % "0.1.0"
+libraryDependencies += "dev.zio" %% "zio-redis" % "<version>"
 ```
 
 ## Example
@@ -32,7 +32,7 @@ To run this example we should put following dependencies in our `build.sbt` file
 
 ```scala
 libraryDependencies ++= Seq(
-  "dev.zio" %% "zio-redis" % "0.1.0",
+  "dev.zio" %% "zio-redis" % "<version>",
   "dev.zio" %% "zio-schema-protobuf" % "0.4.9"
 )
 ```
@@ -61,8 +61,7 @@ object ZIORedisExample extends ZIOAppDefault {
 
   override def run = myApp.provide(
     Redis.layer,
-    RedisExecutor.layer,
-    ZLayer.succeed(RedisConfig.Default),
+    SingleNodeExecutor.local,
     ZLayer.succeed[CodecSupplier](ProtobufCodecSupplier)
   )
 }
@@ -73,7 +72,7 @@ object ZIORedisExample extends ZIOAppDefault {
 To test you can use the embedded redis instance by adding to your build:
 
 ```scala
-libraryDependencies := "dev.zio" %% "zio-redis-embedded" % "0.1.0"
+libraryDependencies := "dev.zio" %% "zio-redis-embedded" % "<version>"
 ```
 
 Then you can supply `EmbeddedRedis.layer.orDie` as your `RedisConfig` and you're good to go!
@@ -108,8 +107,8 @@ object EmbeddedRedisSpec extends ZIOSpecDefault {
       } yield assert(found)(isSome(equalTo(item)))
     }
   ).provideShared(
-    EmbeddedRedis.layer.orDie,
-    RedisExecutor.layer.orDie,
+    EmbeddedRedis.layer,
+    SingleNodeExecutor.layer,
     ZLayer.succeed[CodecSupplier](ProtobufCodecSupplier),
     Redis.layer
   ) @@ TestAspect.silentLogging

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

@@ -25,7 +25,7 @@ import zio.schema.codec.{BinaryCodec, ProtobufCodec}
 trait BenchmarkRuntime {
   final def execute(query: ZIO[Redis, RedisError, Unit]): Unit =
     Unsafe.unsafe { implicit unsafe =>
-      zio.Runtime.default.unsafe.run(query.provideLayer(BenchmarkRuntime.Layer)).getOrThrowFiberFailure()
+      Runtime.default.unsafe.run(query.provideLayer(BenchmarkRuntime.Layer)).getOrThrowFiberFailure()
     }
 
   final def execute[Client: QueryRunner](query: Client => CIO[Unit]): Unit =
@@ -35,7 +35,7 @@ trait BenchmarkRuntime {
 object BenchmarkRuntime {
   private final val Layer =
     ZLayer.make[Redis](
-      RedisExecutor.local,
+      SingleNodeExecutor.local,
       ZLayer.succeed[CodecSupplier](new CodecSupplier {
         def get[A: Schema]: BinaryCodec[A] = ProtobufCodec.protobufCodec
       }),

docs/index.md

@@ -61,8 +61,7 @@ object ZIORedisExample extends ZIOAppDefault {
 
   override def run = myApp.provide(
     Redis.layer,
-    RedisExecutor.layer,
-    ZLayer.succeed(RedisConfig.Default),
+    SingleNodeExecutor.local,
     ZLayer.succeed[CodecSupplier](ProtobufCodecSupplier)
   )
 }
@@ -108,8 +107,8 @@ object EmbeddedRedisSpec extends ZIOSpecDefault {
       } yield assert(found)(isSome(equalTo(item)))
     }
   ).provideShared(
-    EmbeddedRedis.layer.orDie,
-    RedisExecutor.layer.orDie,
+    EmbeddedRedis.layer,
+    SingleNodeExecutor.layer,
     ZLayer.succeed[CodecSupplier](ProtobufCodecSupplier),
     Redis.layer
   ) @@ TestAspect.silentLogging

embedded/src/main/scala/zio/redis/embedded/EmbeddedRedis.scala

@@ -34,14 +34,15 @@ object EmbeddedRedis {
       findFreePort
     }
 
-  val layer: ZLayer[Any, Throwable, RedisConfig] = ZLayer.scoped(
-    for {
-      port <- findFreePort
-      redisServer <- ZIO.acquireRelease(ZIO.attemptBlockingIO(new RedisServer(port)))(redisServer =>
-                       ZIO.attemptBlockingIO(redisServer.stop).ignoreLogged
-                     )
-      _ <- ZIO.attemptBlockingIO(redisServer.start)
-    } yield RedisConfig("localhost", port)
-  )
+  lazy val layer: ZLayer[Any, Throwable, RedisConfig] =
+    ZLayer.scoped {
+      for {
+        port <- findFreePort
+        redisServer <- ZIO.acquireRelease(ZIO.attemptBlockingIO(new RedisServer(port)))(redisServer =>
+                         ZIO.attemptBlockingIO(redisServer.stop).ignoreLogged
+                       )
+        _ <- ZIO.attemptBlockingIO(redisServer.start)
+      } yield RedisConfig("localhost", port)
+    }
 
 }

embedded/src/test/scala/zio/redis/embedded/EmbeddedRedisSpec.scala

@@ -42,8 +42,8 @@ object EmbeddedRedisSpec extends ZIOSpecDefault {
       } yield assert(found)(isSome(equalTo(item)))
     }
   ).provideShared(
-    EmbeddedRedis.layer.orDie,
-    RedisExecutor.layer.orDie,
+    EmbeddedRedis.layer,
+    SingleNodeExecutor.layer,
     ZLayer.succeed[CodecSupplier](new CodecSupplier {
       def get[A: Schema]: BinaryCodec[A] = ProtobufCodec.protobufCodec
     }),

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.{CodecSupplier, Redis, RedisExecutor}
+import zio.redis._
 import zio.schema.Schema
 import zio.schema.codec.{BinaryCodec, ProtobufCodec}
 
@@ -33,8 +33,8 @@ object Main extends ZIOAppDefault {
         AppConfig.layer,
         ContributorsCache.layer,
         HttpClientZioBackend.layer(),
-        RedisExecutor.layer,
         Redis.layer,
+        SingleNodeExecutor.layer,
         ZLayer.succeed[CodecSupplier](new CodecSupplier {
           def get[A: Schema]: BinaryCodec[A] = ProtobufCodec.protobufCodec
         })

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

@@ -19,12 +19,13 @@ package zio.redis
 import zio._
 import zio.redis.ClusterExecutor._
 import zio.redis.api.Cluster.AskingCommand
+import zio.redis.internal.{RedisConnection, RespCommand, RespCommandArgument, RespValue}
 import zio.redis.options.Cluster._
 
 import java.io.IOException
 
-final case class ClusterExecutor(
-  clusterConnectionRef: Ref.Synchronized[ClusterConnection],
+final class ClusterExecutor private (
+  clusterConnection: Ref.Synchronized[ClusterConnection],
   config: RedisClusterConfig,
   scope: Scope.Closeable
 ) extends RedisExecutor {
@@ -56,26 +57,26 @@ final case class ClusterExecutor(
     }
 
     for {
-      keyOpt <- ZIO.succeed(command.args.collectFirst { case key: RespArgument.Key => key })
+      keyOpt <- ZIO.succeed(command.args.collectFirst { case key: RespCommandArgument.Key => key })
       keySlot = keyOpt.fold(Slot.Default)(key => Slot((key.asCRC16 & (SlotsAmount - 1)).toLong))
       result <- executeSafe(keySlot)
     } yield result
   }
 
   private def executor(slot: Slot): IO[RedisError.IOError, RedisExecutor] =
-    clusterConnectionRef.get.map(_.executor(slot)).flatMap(ZIO.fromOption(_).orElseFail(CusterKeyExecutorError))
+    clusterConnection.get.map(_.executor(slot)).flatMap(ZIO.fromOption(_).orElseFail(CusterKeyExecutorError))
 
   // TODO introduce max connection amount
   private def executor(address: RedisUri): IO[RedisError.IOError, RedisExecutor] =
-    clusterConnectionRef.modifyZIO { cc =>
+    clusterConnection.modifyZIO { cc =>
       val executorOpt = cc.executors.get(address).map(es => (es.executor, cc))
       val enrichedClusterIO =
         scope.extend[Any](connectToNode(address)).map(es => (es.executor, cc.addExecutor(address, es)))
       ZIO.fromOption(executorOpt).catchAll(_ => enrichedClusterIO)
     }
 
   private def refreshConnect: IO[RedisError, Unit] =
-    clusterConnectionRef.updateZIO { connection =>
+    clusterConnection.updateZIO { connection =>
       val addresses = connection.partitions.flatMap(_.addresses)
       for {
         cluster <- scope.extend[Any](initConnectToCluster(addresses))
@@ -110,11 +111,11 @@ object ClusterExecutor {
     scope: Scope.Closeable
   ): ZIO[Scope, RedisError, ClusterExecutor] =
     for {
-      clusterConnection    <- initConnectToCluster(config.addresses)
-      clusterConnectionRef <- Ref.Synchronized.make(clusterConnection)
-      clusterExec           = ClusterExecutor(clusterConnectionRef, config, scope)
-      _                    <- logScopeFinalizer("Cluster executor is closed")
-    } yield clusterExec
+      connection <- initConnectToCluster(config.addresses)
+      ref        <- Ref.Synchronized.make(connection)
+      executor    = new ClusterExecutor(ref, config, scope)
+      _          <- logScopeFinalizer("Cluster executor is closed")
+    } yield executor
 
   private def initConnectToCluster(addresses: Chunk[RedisUri]): ZIO[Scope, RedisError, ClusterConnection] =
     ZIO
@@ -141,15 +142,15 @@ object ClusterExecutor {
   private def connectToNode(address: RedisUri) =
     for {
       closableScope <- Scope.make
-      connection    <- closableScope.extend[Any](RedisConnectionLive.create(RedisConfig(address.host, address.port)))
+      connection    <- closableScope.extend[Any](RedisConnection.create(RedisConfig(address.host, address.port)))
       executor      <- closableScope.extend[Any](SingleNodeExecutor.create(connection))
       layerScope    <- ZIO.scope
       _             <- layerScope.addFinalizerExit(closableScope.close(_))
     } yield ExecutorScope(executor, closableScope)
 
   private def redis(address: RedisUri) = {
-    val executorLayer = ZLayer.succeed(RedisConfig(address.host, address.port)) >>> RedisExecutor.layer
-    val codecLayer    = ZLayer.succeed[CodecSupplier](CodecSupplier.utf8string)
+    val executorLayer = ZLayer.succeed(RedisConfig(address.host, address.port)) >>> SingleNodeExecutor.layer
+    val codecLayer    = ZLayer.succeed[CodecSupplier](CodecSupplier.utf8)
     val redisLayer    = executorLayer ++ codecLayer >>> Redis.layer
     for {
       closableScope <- Scope.make

redis/src/main/scala/zio/redis/CodecSupplier.scala

@@ -16,7 +16,7 @@
 
 package zio.redis
 
-import zio.redis.codecs.StringUtf8Codec
+import zio.redis.internal.Utf8Codec
 import zio.schema.Schema
 import zio.schema.codec.BinaryCodec
 
@@ -25,7 +25,7 @@ trait CodecSupplier {
 }
 
 object CodecSupplier {
-  def utf8string: CodecSupplier = new CodecSupplier {
-    def get[A: Schema]: BinaryCodec[A] = StringUtf8Codec.codec
+  def utf8: CodecSupplier = new CodecSupplier {
+    def get[A: Schema]: BinaryCodec[A] = Utf8Codec.codec
   }
 }