Use existing codecs in tests and examples (#557)

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

@@ -20,8 +20,7 @@ import cats.effect.{IO => CIO}
 import zio.internal.Platform
 import zio.logging.Logging
 import zio.redis._
-import zio.redis.codec.StringUtf8Codec
-import zio.schema.codec.Codec
+import zio.schema.codec.{Codec, ProtobufCodec}
 import zio.{BootstrapRuntime, Has, ZIO, ZLayer}
 
 trait BenchmarkRuntime extends BootstrapRuntime {
@@ -37,6 +36,6 @@ trait BenchmarkRuntime extends BootstrapRuntime {
 object BenchmarkRuntime {
   private final val Layer = {
     val executor = Logging.ignore >>> RedisExecutor.local.orDie
-    executor ++ ZLayer.succeed[Codec](StringUtf8Codec) >>> Redis.live
+    executor ++ ZLayer.succeed[Codec](ProtobufCodec) >>> Redis.live
   }
 }

build.sbt

@@ -37,11 +37,12 @@ lazy val redis =
     .settings(buildInfoSettings("zio.redis"))
     .settings(
       libraryDependencies ++= Seq(
-        "dev.zio" %% "zio-streams"  % Zio,
-        "dev.zio" %% "zio-logging"  % "0.5.14",
-        "dev.zio" %% "zio-schema"   % "0.1.1",
-        "dev.zio" %% "zio-test"     % Zio % Test,
-        "dev.zio" %% "zio-test-sbt" % Zio % Test
+        "dev.zio" %% "zio-streams"         % Zio,
+        "dev.zio" %% "zio-logging"         % "0.5.14",
+        "dev.zio" %% "zio-schema"          % "0.1.8",
+        "dev.zio" %% "zio-schema-protobuf" % "0.1.8" % Test,
+        "dev.zio" %% "zio-test"            % Zio     % Test,
+        "dev.zio" %% "zio-test-sbt"        % Zio     % Test
       ),
       testFrameworks := Seq(new TestFramework("zio.test.sbt.ZTestFramework"))
     )
@@ -55,9 +56,10 @@ lazy val benchmarks =
     .settings(
       publish / skip := true,
       libraryDependencies ++= Seq(
-        "dev.profunktor"    %% "redis4cats-effects" % "1.1.1",
-        "io.chrisdavenport" %% "rediculous"         % "0.1.1",
-        "io.laserdisc"      %% "laserdisc-fs2"      % "0.5.0"
+        "dev.profunktor"    %% "redis4cats-effects"  % "1.1.1",
+        "io.chrisdavenport" %% "rediculous"          % "0.1.1",
+        "io.laserdisc"      %% "laserdisc-fs2"       % "0.5.0",
+        "dev.zio"           %% "zio-schema-protobuf" % "0.1.8"
       )
     )
 
@@ -74,7 +76,7 @@ lazy val example =
         "dev.zio"                       %% "zio-streams"                    % Zio,
         "dev.zio"                       %% "zio-config-magnolia"            % "2.0.0",
         "dev.zio"                       %% "zio-config-typesafe"            % "2.0.0",
-        "dev.zio"                       %% "zio-prelude"                    % "1.0.0-RC8",
+        "dev.zio"                       %% "zio-schema-protobuf"            % "0.1.8",
         "dev.zio"                       %% "zio-json"                       % "0.1.5",
         "io.d11"                        %% "zhttp"                          % "1.0.0.0-RC24",
         "io.github.kitlangton"          %% "zio-magic"                      % "0.3.11"

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

@@ -29,9 +29,8 @@ import zio.config.typesafe.TypesafeConfig
 import zio.console._
 import zio.logging.Logging
 import zio.magic._
-import zio.redis.codec.StringUtf8Codec
 import zio.redis.{Redis, RedisExecutor}
-import zio.schema.codec.Codec
+import zio.schema.codec.{Codec, ProtobufCodec}
 
 object Main extends App {
   private val config =
@@ -40,7 +39,7 @@ object Main extends App {
   private val serverConfig = config.narrow(_.server)
   private val redisConfig  = config.narrow(_.redis)
 
-  private val codec         = ZLayer.succeed[Codec](StringUtf8Codec)
+  private val codec         = ZLayer.succeed[Codec](ProtobufCodec)
   private val redisExecutor = Logging.ignore ++ redisConfig >>> RedisExecutor.live
   private val redis         = redisExecutor ++ codec >>> Redis.live
   private val sttp          = AsyncHttpClientZioBackend.layer()

redis/src/main/scala/zio/redis/RedisCommand.scala

@@ -22,11 +22,12 @@ import zio.redis.Input.{StringInput, Varargs}
 final class RedisCommand[-In, +Out] private (val name: String, val input: Input[In], val output: Output[Out]) {
   private[redis] def run(in: In): ZIO[Has[Redis], RedisError, Out] =
     ZIO
-      .serviceWith[Redis] { service =>
-        val executor = service.executor
-        val codec    = service.codec
-        val command  = Varargs(StringInput).encode(name.split(" "))(codec) ++ input.encode(in)(codec)
-        executor.execute(command).flatMap[Any, Throwable, Out](out => ZIO.effect(output.unsafeDecode(out)(codec)))
+      .serviceWith[Redis] { redis =>
+        val command = Varargs(StringInput).encode(name.split(" "))(redis.codec) ++ input.encode(in)(redis.codec)
+
+        redis.executor
+          .execute(command)
+          .flatMap[Any, Throwable, Out](out => ZIO(output.unsafeDecode(out)(redis.codec)))
       }
       .refineToOrDie[RedisError]
 }

redis/src/main/scala/zio/redis/RespValue.scala

@@ -23,12 +23,12 @@ import java.nio.charset.StandardCharsets
 
 sealed trait RespValue extends Product with Serializable { self =>
   import RespValue._
-  import RespValue.internal.{Headers, NullStrEncoded, NullArrEncoded, CrLf}
+  import RespValue.internal.{Headers, NullStringEncoded, NullArrayEncoded, CrLf}
 
   final def serialize: Chunk[Byte] =
     self match {
-      case NullBulkString  => NullStrEncoded
-      case NullArray       => NullArrEncoded
+      case NullBulkString  => NullStringEncoded
+      case NullArray       => NullArrayEncoded
       case SimpleString(s) => Headers.SimpleString +: encode(s)
       case Error(s)        => Headers.Error +: encode(s)
       case Integer(i)      => Headers.Integer +: encode(i.toString)
@@ -102,11 +102,11 @@ object RespValue {
       final val Array: Byte        = '*'
     }
 
-    final val CrLf: Chunk[Byte]           = Chunk('\r', '\n')
-    final val NullStringPrefix: String    = "$-1"
-    final val NullArrayPrefix: String     = "*-1"
-    final val NullStrEncoded: Chunk[Byte] = Chunk.fromArray("$-1\r\n".getBytes(StandardCharsets.US_ASCII))
-    final val NullArrEncoded: Chunk[Byte] = Chunk.fromArray("*-1\r\n".getBytes(StandardCharsets.US_ASCII))
+    final val CrLf: Chunk[Byte]              = Chunk('\r', '\n')
+    final val NullArrayEncoded: Chunk[Byte]  = Chunk.fromArray("*-1\r\n".getBytes(StandardCharsets.US_ASCII))
+    final val NullArrayPrefix: String        = "*-1"
+    final val NullStringEncoded: Chunk[Byte] = Chunk.fromArray("$-1\r\n".getBytes(StandardCharsets.US_ASCII))
+    final val NullStringPrefix: String       = "$-1"
 
     sealed trait State { self =>
       import State._

redis/src/main/scala/zio/redis/codec/StringUtf8Codec.scala

-/*
- * Copyright 2021 John A. De Goes and the ZIO contributors
- *
- * 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 zio.redis.codec
-
-import zio.redis.RedisError.CodecError
-import zio.schema.Schema
-import zio.schema.StandardType.{DoubleType, IntType, LongType}
-import zio.schema.codec.Codec
-import zio.stream.ZTransducer
-import zio.{Chunk, ZIO}
-
-import java.nio.charset.StandardCharsets
-
-object StringUtf8Codec extends Codec {
-  def encoder[A](schema: Schema[A]): ZTransducer[Any, Nothing, A, Byte] =
-    ZTransducer.fromPush { (opt: Option[Chunk[A]]) =>
-      ZIO.succeed(opt.map(values => values.flatMap(Encoder.encode(schema, _))).getOrElse(Chunk.empty))
-    }
-
-  def encode[A](schema: Schema[A]): A => Chunk[Byte] = { a =>
-    Encoder.encode(schema, a)
-  }
-
-  def decoder[A](schema: Schema[A]): ZTransducer[Any, String, Byte, A] =
-    ZTransducer.fromPush { (opt: Option[Chunk[Byte]]) =>
-      ZIO.fromEither(opt.map(chunk => Decoder.decode(schema, chunk).map(Chunk(_))).getOrElse(Right(Chunk.empty)))
-    }
-
-  def decode[A](schema: Schema[A]): Chunk[Byte] => Either[String, A] = { ch =>
-    Decoder.decode(schema, ch)
-  }
-
-  object Encoder {
-    def encode[A](schema: Schema[A], value: A): Chunk[Byte] =
-      schema match {
-        case Schema.Primitive(_) => Chunk.fromArray(value.toString.getBytes(StandardCharsets.UTF_8))
-        case _                   => throw CodecError("the codec support only primitives")
-      }
-  }
-
-  object Decoder {
-    def decode[A](schema: Schema[A], chunk: Chunk[Byte]): Either[String, A] = {
-      def utf8String = new String(chunk.toArray, StandardCharsets.UTF_8)
-
-      schema match {
-        case Schema.Primitive(IntType)    => Right(utf8String.toInt.asInstanceOf[A])
-        case Schema.Primitive(LongType)   => Right(utf8String.toLong.asInstanceOf[A])
-        case Schema.Primitive(DoubleType) => Right(utf8String.toDouble.asInstanceOf[A])
-        case Schema.Primitive(_)          => Right(utf8String.asInstanceOf[A])
-        case _                            => Left("the codec support only primitives")
-      }
-    }
-  }
-}

redis/src/test/scala/zio/redis/ApiSpec.scala

@@ -34,11 +34,7 @@ object ApiSpec
         hashSuite,
         streamsSuite,
         scriptingSpec
-      ).provideCustomLayerShared {
-        val executor = Logging.ignore >>> RedisExecutor.local.orDie
-        val redis    = executor ++ ZLayer.succeed(codec) >>> Redis.live
-        redis ++ Clock.live
-      } @@ sequential,
+      ).provideCustomLayerShared(LiveLayer) @@ sequential,
       suite("Test Executor")(
         connectionSuite,
         keysSuite,
@@ -49,8 +45,17 @@ object ApiSpec
         sortedSetsSuite,
         geoSuite,
         stringsSuite
-      ).filterAnnotations(TestAnnotation.tagged)(t => !t.contains(TestExecutorUnsupportedTag))
+      ).filterAnnotations(TestAnnotation.tagged)(t => !t.contains(BaseSpec.TestExecutorUnsupported))
         .get
-        .provideSomeLayer[TestEnvironment](RedisExecutor.test ++ ZLayer.succeed(codec) >>> Redis.live)
+        .provideSomeLayer[TestEnvironment](TestLayer)
     )
+
+  private val LiveLayer = {
+    val executor = Logging.ignore >>> RedisExecutor.local.orDie
+    val redis    = executor ++ ZLayer.succeed(codec) >>> Redis.live
+    redis ++ Clock.live
+  }
+
+  private val TestLayer =
+    RedisExecutor.test ++ ZLayer.succeed(codec) >>> Redis.live
 }

redis/src/test/scala/zio/redis/BaseSpec.scala

@@ -3,8 +3,7 @@ package zio.redis
 import zio.UIO
 import zio.duration._
 import zio.random.Random
-import zio.redis.codec.StringUtf8Codec
-import zio.schema.codec.Codec
+import zio.schema.codec.{Codec, ProtobufCodec}
 import zio.test.TestAspect.tag
 import zio.test._
 import zio.test.environment.Live
@@ -13,19 +12,28 @@ import java.time.Instant
 import java.util.UUID
 
 trait BaseSpec extends DefaultRunnableSpec {
+  implicit val codec: Codec = ProtobufCodec
+
   override def aspects: List[TestAspectAtLeastR[Live]] = List(TestAspect.timeout(60.seconds))
 
   def instantOf(millis: Long): UIO[Instant] = UIO(Instant.now().plusMillis(millis))
 
-  implicit val codec: Codec = StringUtf8Codec
+  final val genStringRedisTypeOption: Gen[Random, Option[RedisType]] =
+    Gen.option(Gen.constSample(Sample.noShrink(RedisType.String)))
 
-  val uuid: UIO[String] = UIO(UUID.randomUUID().toString)
+  final val genCountOption: Gen[Random, Option[Count]] =
+    Gen.option(Gen.long(0, 100000).map(Count))
 
-  val TestExecutorUnsupportedTag                   = "test executor unsupported"
-  lazy val testExecutorUnsupported: TestAspectPoly = tag(TestExecutorUnsupportedTag)
+  final val genPatternOption: Gen[Random, Option[String]] =
+    Gen.option(Gen.constSample(Sample.noShrink("*")))
 
-  val genStringRedisTypeOption: Gen[Random, Option[RedisType]] =
-    Gen.option(Gen.constSample(Sample.noShrink(RedisType.String)))
-  val genCountOption: Gen[Random, Option[Count]]    = Gen.option(Gen.long(0, 100000).map(Count))
-  val genPatternOption: Gen[Random, Option[String]] = Gen.option(Gen.constSample(Sample.noShrink("*")))
+  final val uuid: UIO[String] =
+    UIO(UUID.randomUUID().toString)
+
+  final val testExecutorUnsupported: TestAspectPoly =
+    tag(BaseSpec.TestExecutorUnsupported)
+}
+
+object BaseSpec {
+  final val TestExecutorUnsupported = "test executor unsupported"
 }

redis/src/test/scala/zio/redis/KeysSpec.scala

@@ -5,8 +5,7 @@ import zio.duration._
 import zio.logging.Logging
 import zio.random.Random
 import zio.redis.RedisError.ProtocolError
-import zio.redis.codec.StringUtf8Codec
-import zio.schema.codec.Codec
+import zio.schema.codec.{Codec, ProtobufCodec}
 import zio.test.Assertion._
 import zio.test.TestAspect._
 import zio.test._
@@ -389,25 +388,25 @@ trait KeysSpec extends BaseSpec {
             _      <- lPush(key, value)
             prefix <- uuid
             _      <- set(s"${prefix}_$value", "A")
-            sorted <- sort(key, get = Some((s"${prefix}_*", List.empty)), alpha = Some(Alpha)).returning[String]
+            sorted <- sort(key, get = Some(s"${prefix}_*" -> List.empty), alpha = Some(Alpha)).returning[String]
           } yield assert(sorted)(equalTo(Chunk("A")))
         },
-        testM("getting multiple value referenced by a key-value pair") {
+        testM("getting multiple values referenced by a key-value pair") {
           for {
             key     <- uuid
-            value1   = 1
-            value2   = 2
+            value1  <- uuid
+            value2  <- uuid
             _       <- lPush(key, value1, value2)
             prefix  <- uuid
             _       <- set(s"${prefix}_$value1", "A1")
             _       <- set(s"${prefix}_$value2", "A2")
             prefix2 <- uuid
-            _       <- set(s"${prefix2}_$value1", "01")
-            _       <- set(s"${prefix2}_$value2", "02")
+            _       <- set(s"${prefix2}_$value1", "B1")
+            _       <- set(s"${prefix2}_$value2", "B2")
             sorted <- sort(key, get = Some((s"${prefix}_*", List(s"${prefix2}_*"))), alpha = Some(Alpha))
                         .returning[String]
-          } yield assert(sorted)(equalTo(Chunk("A1", "01", "A2", "02")))
-        },
+          } yield assert(sorted)(equalTo(Chunk("A1", "B1", "A2", "B2")))
+        } @@ flaky,
         testM("sort and store result") {
           for {
             key       <- uuid
@@ -427,6 +426,6 @@ object KeysSpec {
 
   final val SecondExecutor: Layer[RedisError.IOError, Has[Redis]] = {
     val executor = Logging.ignore ++ ZLayer.succeed(RedisConfig("localhost", 6380)) >>> RedisExecutor.live
-    (executor ++ ZLayer.succeed[Codec](StringUtf8Codec) >>> Redis.live).fresh
+    (executor ++ ZLayer.succeed[Codec](ProtobufCodec) >>> Redis.live).fresh
   }
 }