#998 Client side caching

Author: a-khakimov

build.sbt

@@ -44,6 +44,7 @@ val commonSettings = Seq(
   testFrameworks += new TestFramework("munit.Framework"),
   libraryDependencies ++= Seq(
     Libraries.catsEffectKernel,
+    Libraries.catsEffectStd,
     Libraries.redisClient,
     Libraries.catsEffect      % Test,
     Libraries.catsLaws        % Test,

modules/core/src/main/scala/dev/profunktor/redis4cats/caching/CacheAccessor.scala

@@ -0,0 +1,23 @@
+/*
+ * Copyright 2018-2021 ProfunKtor
+ *
+ * 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 dev.profunktor.redis4cats.caching
+
+trait CacheAccessor[F[_], K, V] {
+  def get(key: K): F[V]
+  def put(key: K, value: V): F[Unit]
+  def evict(key: K): F[Unit]
+}

modules/core/src/main/scala/dev/profunktor/redis4cats/caching/ClientSideCaching.scala

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2018-2021 ProfunKtor
+ *
+ * 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 dev.profunktor.redis4cats.caching
+
+import cats.effect.{ ContextShift, Effect, Resource, Sync }
+import dev.profunktor.redis4cats.effect.RedisExecutor
+import io.lettuce.core.TrackingArgs
+import io.lettuce.core.api.StatefulRedisConnection
+import io.lettuce.core.support.caching.{
+  CacheAccessor => JCacheAccessor,
+  CacheFrontend => JCacheFrontend,
+  ClientSideCaching => JClientSideCaching
+}
+
+object ClientSideCaching {
+
+  def make[F[_]: ContextShift: Sync: Effect, K, V](
+      connection: StatefulRedisConnection[K, V],
+      args: TrackingArgs,
+      cacheAccessor: CacheAccessor[F, K, V]
+  ): Resource[F, JCacheFrontend[K, V]] =
+    RedisExecutor.make[F].flatMap { redisExecutor =>
+      Resource.make[F, JCacheFrontend[K, V]] {
+        Sync[F].delay {
+          JClientSideCaching.enable(
+            new JCacheAccessor[K, V] {
+              override def get(key: K): V              = redisExecutor.unsafeRun(cacheAccessor.get(key))
+              override def put(key: K, value: V): Unit = redisExecutor.unsafeRun(cacheAccessor.put(key, value))
+              override def evict(key: K): Unit         = redisExecutor.unsafeRun(cacheAccessor.evict(key))
+            },
+            connection,
+            args
+          )
+        }
+      }(cacheFrontend => Sync[F].delay(cacheFrontend.close()))
+    }
+}

modules/core/src/main/scala/dev/profunktor/redis4cats/effect/TxExecutor.scala

@@ -29,26 +29,30 @@ import scala.util.control.NonFatal
 
 import cats.effect.kernel._
 import cats.syntax.all._
+import cats.effect.std.Dispatcher
 
 private[redis4cats] trait TxExecutor[F[_]] {
   def delay[A](thunk: => A): F[A]
   def eval[A](fa: F[A]): F[A]
   def start[A](fa: F[A]): F[Fiber[F, Throwable, A]]
-  def liftK[G[_]: Async]: TxExecutor[G]
+  def liftK[G[_]: Async: Dispatcher]: TxExecutor[G]
+  def unsafeRun[A](fa: F[A]): A
 }
 
 private[redis4cats] object TxExecutor {
   def make[F[_]: Async]: Resource[F, TxExecutor[F]] =
-    Resource
-      .make(Sync[F].delay(Executors.newFixedThreadPool(1, TxThreadFactory))) { ec =>
-        Sync[F]
-          .delay(ec.shutdownNow())
-          .ensure(new IllegalStateException("There were outstanding tasks at time of shutdown of the Redis thread"))(
-            _.isEmpty
-          )
-          .void
-      }
-      .map(es => fromEC(exitOnFatal(ExecutionContext.fromExecutorService(es))))
+    Dispatcher.parallel[F].flatMap { dispatcher =>
+      Resource
+        .make(Sync[F].delay(Executors.newFixedThreadPool(1, TxThreadFactory))) { ec =>
+          Sync[F]
+            .delay(ec.shutdownNow())
+            .ensure(new IllegalStateException("There were outstanding tasks at time of shutdown of the Redis thread"))(
+              _.isEmpty
+            )
+            .void
+        }
+        .map(es => fromEC(exitOnFatal(ExecutionContext.fromExecutorService(es)))(dispatcher))
+    }
 
   private def exitOnFatal(ec: ExecutionContext): ExecutionContext = new ExecutionContext {
     def execute(r: Runnable): Unit =
@@ -70,11 +74,12 @@ private[redis4cats] object TxExecutor {
       ec.reportFailure(t)
   }
 
-  private def fromEC[F[_]: Async](ec: ExecutionContext): TxExecutor[F] =
+  private def fromEC[F[_]: Async](ec: ExecutionContext)(dispatcher: Dispatcher[F]): TxExecutor[F] =
     new TxExecutor[F] {
       def delay[A](thunk: => A): F[A]                   = eval(Sync[F].delay(thunk))
       def eval[A](fa: F[A]): F[A]                       = Async[F].evalOn(fa, ec)
       def start[A](fa: F[A]): F[Fiber[F, Throwable, A]] = Async[F].startOn(fa, ec)
-      def liftK[G[_]: Async]: TxExecutor[G]             = fromEC[G](ec)
+      def liftK[G[_]: Async: Dispatcher]: TxExecutor[G] = fromEC[G](ec)(implicitly)
+      def unsafeRun[A](fa: F[A]): A                     = dispatcher.unsafeRunSync(fa)
     }
 }

project/Dependencies.scala

@@ -26,6 +26,7 @@ object Dependencies {
     def log4cats(artifact: String): ModuleID = "org.typelevel" %% s"log4cats-$artifact" % V.log4cats
 
     val catsEffectKernel = "org.typelevel" %% "cats-effect-kernel" % V.catsEffect
+    val catsEffectStd    = "org.typelevel" %% "cats-effect-std"    % V.catsEffect
     val fs2Core          = "co.fs2"        %% "fs2-core"           % V.fs2
     val keyPool          = "org.typelevel" %% "keypool"            % V.keyPool