Scripting API (scriptDebug, scriptFlush, scriptKill)

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

@@ -36,6 +36,6 @@ object BenchmarkRuntime {
     ZLayer.make[Redis](
       RedisExecutor.local,
       ZLayer.succeed[BinaryCodec](ProtobufCodec),
-      RedisLive.layer
+      Redis.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.{Redis, RedisExecutor}
 import zio.schema.codec.{BinaryCodec, ProtobufCodec}
 
 object Main extends ZIOAppDefault {
@@ -33,7 +33,7 @@ object Main extends ZIOAppDefault {
         ContributorsCache.layer,
         HttpClientZioBackend.layer(),
         RedisExecutor.layer,
-        RedisLive.layer,
+        Redis.layer,
         ZLayer.succeed[BinaryCodec](ProtobufCodec)
       )
       .exitCode

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

@@ -152,7 +152,7 @@ object ClusterExecutor {
   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 redisLayer    = executorLayer ++ codecLayer >>> Redis.layer
     for {
       closableScope <- Scope.make
       layer         <- closableScope.extend[Any](redisLayer.memoize)

redis/src/main/scala/zio/redis/Input.scala

@@ -597,6 +597,11 @@ object Input {
       Chunk.single(encodeString(data.stringify))
   }
 
+  case object ScriptFlushInput extends Input[FlushMode] {
+    def encode(data: FlushMode)(implicit codec: BinaryCodec): Chunk[BulkString] =
+      Chunk.single(encodeString(data.stringify))
+  }
+
   case object WithScoresInput extends Input[WithScores] {
     def encode(data: WithScores)(implicit codec: BinaryCodec): Chunk[RespValue.BulkString] =
       Chunk.single(encodeString(data.stringify))

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

@@ -36,9 +36,13 @@ trait Redis
   def executor: RedisExecutor
 }
 
-final case class RedisLive(codec: BinaryCodec, executor: RedisExecutor) extends Redis
-
-object RedisLive {
+object Redis {
   lazy val layer: URLayer[RedisExecutor with BinaryCodec, Redis] =
-    ZLayer.fromFunction(RedisLive.apply _)
+    ZLayer.fromZIO(for {
+      redisExecutor <- ZIO.service[RedisExecutor]
+      binaryCodec   <- ZIO.service[BinaryCodec]
+    } yield new Redis {
+      def codec: BinaryCodec      = binaryCodec
+      def executor: RedisExecutor = redisExecutor
+    })
 }

redis/src/main/scala/zio/redis/api/Scripting.scala

@@ -74,6 +74,19 @@ trait Scripting extends RedisEnvironment {
     }
   }
 
+  /**
+   * Set the debug mode for executed scripts.
+   *
+   * @param mode
+   *   mode in which scripts debug is going to work ["YES", "SYNC", "NO"]
+   * @return
+   *   the Unit value.
+   */
+  def scriptDebug(mode: DebugMode): IO[RedisError, Unit] = {
+    val command = RedisCommand(ScriptDebug, ScriptDebugInput, UnitOutput, codec, executor)
+    command.run(mode)
+  }
+
   /**
    * Checks existence of the scripts in the script cache.
    *
@@ -90,6 +103,31 @@ trait Scripting extends RedisEnvironment {
     command.run((sha1, sha1s.toList))
   }
 
+  /**
+   * Remove all the scripts from the script cache.
+   *
+   * @param mode
+   *   mode in which script flush is going to be executed ["ASYNC", "SYNC"] Note: "SYNC" mode is used by default (if no
+   *   mode is provided)
+   * @return
+   *   the Unit value.
+   */
+  def scriptFlush(mode: Option[FlushMode] = None): IO[RedisError, Unit] = {
+    val command = RedisCommand(ScriptFlush, OptionalInput(ScriptFlushInput), UnitOutput, codec, executor)
+    command.run(mode)
+  }
+
+  /**
+   * Kill the currently executing EVAL script, assuming no write operation was yet performed by the script.
+   *
+   * @return
+   *   the Unit value.
+   */
+  def scriptKill: IO[RedisError, Unit] = {
+    val command = RedisCommand(ScriptKill, NoInput, UnitOutput, codec, executor)
+    command.run(())
+  }
+
   /**
    * Loads a script into the scripts cache. After the script is loaded into the script cache it could be evaluated using
    * the [[zio.redis.api.Scripting.evalSha]] method.
@@ -108,6 +146,9 @@ trait Scripting extends RedisEnvironment {
 private[redis] object Scripting {
   final val Eval         = "EVAL"
   final val EvalSha      = "EVALSHA"
+  final val ScriptDebug  = "SCRIPT DEBUG"
   final val ScriptExists = "SCRIPT EXISTS"
+  final val ScriptFlush  = "SCRIPT FLUSH"
+  final val ScriptKill   = "SCRIPT KILL"
   final val ScriptLoad   = "SCRIPT LOAD"
 }

redis/src/main/scala/zio/redis/options/Scripting.scala

@@ -31,4 +31,17 @@ trait Scripting {
     case object Sync extends DebugMode
     case object No   extends DebugMode
   }
+
+  sealed trait FlushMode { self =>
+    private[redis] final def stringify: String =
+      self match {
+        case FlushMode.Async => "ASYNC"
+        case FlushMode.Sync  => "SYNC"
+      }
+  }
+
+  object FlushMode {
+    case object Async extends FlushMode
+    case object Sync  extends FlushMode
+  }
 }

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

@@ -40,7 +40,7 @@ object ApiSpec
         scriptingSpec
       )
 
-    val Layer: Layer[Any, Redis] = ZLayer.make[Redis](RedisExecutor.local.orDie, ZLayer.succeed(codec), RedisLive.layer)
+    val Layer: Layer[Any, Redis] = ZLayer.make[Redis](RedisExecutor.local.orDie, ZLayer.succeed(codec), Redis.layer)
   }
 
   private object Cluster {
@@ -65,7 +65,7 @@ object ApiSpec
         ZLayer.succeed(RedisClusterConfig(Chunk(RedisUri("localhost", 5000)))),
         ClusterExecutor.layer,
         ZLayer.succeed(codec),
-        RedisLive.layer
+        Redis.layer
       )
   }
 

redis/src/test/scala/zio/redis/ClusterExecutorSpec.scala

@@ -70,7 +70,7 @@ object ClusterExecutorSpec extends BaseSpec {
       ZLayer.succeed(RedisConfig(uri.host, uri.port)),
       RedisExecutor.layer,
       ZLayer.succeed(codec),
-      RedisLive.layer
+      Redis.layer
     )
 
   private val ClusterLayer: Layer[Any, Redis] = {
@@ -80,7 +80,7 @@ object ClusterExecutorSpec extends BaseSpec {
       ZLayer.succeed(RedisClusterConfig(Chunk(address1, address2))),
       ClusterExecutor.layer.orDie,
       ZLayer.succeed(codec),
-      RedisLive.layer
+      Redis.layer
     )
   }
 }

redis/src/test/scala/zio/redis/InputSpec.scala

@@ -817,6 +817,35 @@ object InputSpec extends BaseSpec {
           } yield assert(result)(equalTo(respArgs("4.2", "5.2")))
         }
       ),
+      suite("ScriptDebug")(
+        test("yes") {
+          for {
+            result <- ZIO.attempt(ScriptDebugInput.encode(DebugMode.Yes))
+          } yield assert(result)(equalTo(respArgs("YES")))
+        },
+        test("sync") {
+          for {
+            result <- ZIO.attempt(ScriptDebugInput.encode(DebugMode.Sync))
+          } yield assert(result)(equalTo(respArgs("SYNC")))
+        },
+        test("no") {
+          for {
+            result <- ZIO.attempt(ScriptDebugInput.encode(DebugMode.No))
+          } yield assert(result)(equalTo(respArgs("NO")))
+        }
+      ),
+      suite("ScriptFlush")(
+        test("asynchronous") {
+          for {
+            result <- ZIO.attempt(ScriptFlushInput.encode(FlushMode.Async))
+          } yield assert(result)(equalTo(respArgs("ASYNC")))
+        },
+        test("synchronous") {
+          for {
+            result <- ZIO.attempt(ScriptFlushInput.encode(FlushMode.Sync))
+          } yield assert(result)(equalTo(respArgs("SYNC")))
+        }
+      ),
       suite("String")(
         test("non-empty value") {
           for {

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

@@ -465,7 +465,7 @@ object KeysSpec {
         RedisConnectionLive.layer,
         SingleNodeExecutor.layer,
         ZLayer.succeed[BinaryCodec](ProtobufCodec),
-        RedisLive.layer
+        Redis.layer
       )
       .fresh
 }

redis/src/test/scala/zio/redis/ScriptingSpec.scala

@@ -163,6 +163,26 @@ trait ScriptingSpec extends BaseSpec {
           } yield assert(res)(isLeft(isSubtype[NoScript](hasField("message", _.message, equalTo(error)))))
         }
       ),
+      suite("scriptDebug")(
+        test("enable non-blocking asynchronous debugging") {
+          for {
+            redis <- ZIO.service[Redis]
+            res   <- redis.scriptDebug(DebugMode.Yes)
+          } yield assert(res)(isUnit)
+        },
+        test("enable blocking synchronous debugging") {
+          for {
+            redis <- ZIO.service[Redis]
+            res   <- redis.scriptDebug(DebugMode.Sync)
+          } yield assert(res)(isUnit)
+        },
+        test("disable debug mode") {
+          for {
+            redis <- ZIO.service[Redis]
+            res   <- redis.scriptDebug(DebugMode.No)
+          } yield assert(res)(isUnit)
+        }
+      ),
       suite("scriptExists")(
         test("return true if scripts are found in the cache") {
           val lua1 = """return "1""""
@@ -183,6 +203,49 @@ trait ScriptingSpec extends BaseSpec {
           } yield assertTrue(res == Chunk(false, false))
         }
       ),
+      suite("scriptFlush")(
+        test("flush scripts in default mode") {
+          val lua1 = """return "1""""
+          val lua2 = """return "2""""
+          for {
+            redis <- ZIO.service[Redis]
+            sha1  <- redis.scriptLoad(lua1)
+            sha2  <- redis.scriptLoad(lua2)
+            res   <- redis.scriptFlush()
+            found <- redis.scriptExists(sha1, sha2)
+          } yield assert(res)(isUnit) && assertTrue(found == Chunk(false, false))
+        },
+        test("flush scripts in SYNC mode") {
+          val lua1 = """return "1""""
+          val lua2 = """return "2""""
+          for {
+            redis <- ZIO.service[Redis]
+            sha1  <- redis.scriptLoad(lua1)
+            sha2  <- redis.scriptLoad(lua2)
+            res   <- redis.scriptFlush(mode = Some(FlushMode.Sync))
+            found <- redis.scriptExists(sha1, sha2)
+          } yield assert(res)(isUnit) && assertTrue(found == Chunk(false, false))
+        },
+        test("flush scripts in ASYNC mode") {
+          val lua1 = """return "1""""
+          val lua2 = """return "2""""
+          for {
+            redis <- ZIO.service[Redis]
+            sha1  <- redis.scriptLoad(lua1)
+            sha2  <- redis.scriptLoad(lua2)
+            res   <- redis.scriptFlush(mode = Some(FlushMode.Async))
+            found <- redis.scriptExists(sha1, sha2)
+          } yield assert(res)(isUnit) && assertTrue(found == Chunk(false, false))
+        }
+      ),
+      suite("scriptKill")(
+        test("return NOTBUSY when there is no scripts in execution") {
+          for {
+            redis <- ZIO.service[Redis]
+            res   <- redis.scriptKill.either
+          } yield assert(res)(isLeft(isSubtype[RedisError.NotBusy](anything)))
+        }
+      ),
       suite("scriptLoad")(
         test("return OK") {
           val lua = """return "1""""