Improve Redis example

examples/src/test/scala/JedisExample.scala

@@ -17,6 +17,7 @@
 import scala.concurrent.ExecutionContext
 import scala.concurrent.duration._
 
+import cats.Monad
 import cats.data.NonEmptyList
 import cats.effect._
 import cats.instances.list._
@@ -30,6 +31,7 @@ import org.scalatest.{Matchers, WordSpec}
 import java.io._
 import java.nio.charset.Charset
 import redis.clients.jedis._
+import scala.util.Try
 
 import fetch._
 
@@ -58,83 +60,108 @@ object DataSources {
       _ <- fetchNumber(n)
       u <- HttpExample.fetchUser(n)
     } yield u
+
+  def fetchMulti[F[_]: ConcurrentEffect]: Fetch[F, List[HttpExample.User]] =
+    List(4, 5, 6).traverse(HttpExample.fetchUser[F](_))
 }
 
 object Binary {
+  type ByteArray = Array[Byte]
+
+  def byteOutputStream[F[_]](implicit S: Sync[F]): Resource[F, ByteArrayOutputStream] =
+    Resource.fromAutoCloseable(S.delay(new ByteArrayOutputStream()))
+
+  def byteInputStream[F[_]](bin: ByteArray)(
+      implicit S: Sync[F]): Resource[F, ByteArrayInputStream] =
+    Resource.fromAutoCloseable(S.delay(new ByteArrayInputStream(bin)))
+
+  def outputStream[F[_]](b: ByteArrayOutputStream)(
+      implicit S: Sync[F]): Resource[F, ObjectOutputStream] =
+    Resource.fromAutoCloseable(S.delay(new ObjectOutputStream(b)))
+
+  def inputStream[F[_]](b: ByteArrayInputStream)(
+      implicit S: Sync[F]): Resource[F, ObjectInputStream] =
+    Resource.fromAutoCloseable(S.delay(new ObjectInputStream(b)))
+
   def fromString(s: String): Array[Byte] =
     s.getBytes(Charset.forName("UTF-8"))
 
-  def serialize[F[_]: Sync, A](v: A): F[Array[Byte]] = {
-    val baos = new ByteArrayOutputStream()
-    val oos  = new ObjectOutputStream(baos)
-    Sync[F].bracket(Sync[F].pure(oos))({ in =>
-      Sync[F].pure({
-        in.writeObject(v)
-        in.flush()
-        baos.toByteArray
+  def serialize[F[_], A](obj: A)(
+      implicit S: Sync[F]
+  ): F[ByteArray] = {
+    byteOutputStream
+      .mproduct(outputStream(_))
+      .use({
+        case (byte, out) =>
+          S.delay {
+            out.writeObject(obj)
+            out.flush()
+            byte.toByteArray
+          }
       })
-    })({ in =>
-      Sync[F].delay(in.close())
-    })
   }
 
-  def deserialize[F[_]: Sync, I, A](bs: Option[Array[Byte]]): F[Option[A]] =
-    bs match {
-      case None => Sync[F].pure(None)
-      case Some(raw) => {
-        val bais = new ByteArrayInputStream(raw)
-        val oos  = new ObjectInputStream(bais)
-        Sync[F].bracket(
-          Sync[F].pure((bais, oos))
-        )({
-          case (b, o) =>
-            Sync[F].pure({
-              val res = o.readObject()
-              Option(res.asInstanceOf[A])
-            })
-        })({
-          case (b, o) =>
-            Sync[F].delay({
-              b.close()
-              o.close()
-            })
-        })
-      }
-    }
-
+  def deserialize[F[_], A](bin: ByteArray)(
+      implicit S: Sync[F]
+  ): F[Option[A]] = {
+    byteInputStream(bin)
+      .mproduct(inputStream(_))
+      .use({
+        case (byte, in) =>
+          S.delay {
+            val obj = in.readObject()
+            Try(obj.asInstanceOf[A]).toOption
+          }
+      })
+  }
 }
 
 case class RedisCache[F[_]: Sync](host: String) extends DataCache[F] {
-  private val binaryClient = new BinaryJedis(host)
+  private val pool = new JedisPool(host)
 
-  def client: F[BinaryJedis] =
-    Sync[F].pure(binaryClient)
+  def connection: Resource[F, Jedis] =
+    Resource.fromAutoCloseable(Sync[F].delay(pool.getResource))
 
   private def get(i: Array[Byte]): F[Option[Array[Byte]]] =
-    Sync[F].bracket(client)({ c =>
-      Sync[F].pure(Option(c.get(i)))
-    })({ in =>
-      Sync[F].pure(in.close())
-    })
+    connection.use(c => Sync[F].delay(Option(c.get(i))))
 
   private def set(i: Array[Byte], v: Array[Byte]): F[Unit] =
-    Sync[F].bracket(client)({ c =>
-      Sync[F].pure(c.set(i, v)).void
-    })({ in =>
-      Sync[F].pure(in.close())
-    })
+    connection.use(c => Sync[F].delay(c.set(i, v)).void)
 
-  private def identity[I, A](i: I, data: Data[I, A]): Array[Byte] =
+  private def bulkSet(ivs: List[(Array[Byte], Array[Byte])]): F[Unit] =
+    connection.use(c =>
+      Sync[F].delay({
+        val pipe = c.pipelined
+        ivs.foreach(i => pipe.set(i._1, i._2))
+        pipe.sync
+      }))
+
+  private def cacheId[I, A](i: I, data: Data[I, A]): Array[Byte] =
     Binary.fromString(s"${data.identity} ${i}")
 
-  def lookup[I, A](i: I, data: Data[I, A]): F[Option[A]] =
-    get(identity(i, data)) >>= { case (raw) => Binary.deserialize(raw) }
+  override def lookup[I, A](i: I, data: Data[I, A]): F[Option[A]] =
+    get(cacheId(i, data)) >>= {
+      case None    => Sync[F].pure(None)
+      case Some(r) => Binary.deserialize[F, A](r)
+    }
 
-  def insert[I, A](i: I, v: A, data: Data[I, A]): F[DataCache[F]] =
+  override def insert[I, A](i: I, v: A, data: Data[I, A]): F[DataCache[F]] =
     for {
       s <- Binary.serialize(v)
-      _ <- set(identity(i, data), s)
+      _ <- set(cacheId(i, data), s)
+    } yield this
+
+  override def bulkInsert[I, A](vs: List[(I, A)], data: Data[I, A])(
+      implicit M: Monad[F]
+  ): F[DataCache[F]] =
+    for {
+      bin <- vs.traverse({
+        case (id, v) =>
+          Binary.serialize(v).tupleRight(cacheId(id, data))
+      })
+      _ <- Sync[F].delay(bulkSet(bin))
     } yield this
+
 }
 
 class JedisExample extends WordSpec with Matchers {
@@ -146,7 +173,8 @@ class JedisExample extends WordSpec with Matchers {
   implicit val cs: ContextShift[IO] = IO.contextShift(executionContext)
 
   "We can use a Redis cache" in {
-    val cache                           = RedisCache[IO]("localhost")
+    val cache = RedisCache[IO]("localhost")
+
     val io: IO[(Log, HttpExample.User)] = Fetch.runLog[IO](fetch, cache)
 
     val (log, result) = io.unsafeRunSync
@@ -155,7 +183,26 @@ class JedisExample extends WordSpec with Matchers {
     log.rounds.size shouldEqual 2
 
     val io2: IO[(Log, HttpExample.User)] = Fetch.runLog[IO](fetch, cache)
-    val (log2, result2)                  = io2.unsafeRunSync
+
+    val (log2, result2) = io2.unsafeRunSync
+
+    println(result2)
+    log2.rounds.size shouldEqual 0
+  }
+
+  "We can bulk insert in a Redis cache" in {
+    val cache = RedisCache[IO]("localhost")
+
+    val io: IO[(Log, List[HttpExample.User])] = Fetch.runLog[IO](fetchMulti, cache)
+
+    val (log, result) = io.unsafeRunSync
+
+    println(result)
+    log.rounds.size shouldEqual 1
+
+    val io2: IO[(Log, List[HttpExample.User])] = Fetch.runLog[IO](fetchMulti, cache)
+
+    val (log2, result2) = io2.unsafeRunSync
 
     println(result2)
     log2.rounds.size shouldEqual 0