profunktor.dev/redis4cats

gvolpe: I've been thinking about the discussion with @Kurek_gitlab and I played around with something that would allow to have instantiated connections for transactions / pipelines. What are your thoughts on having something like this:

trait Transactor[F[_], K, V] {
  def transact_(ops: RedisCommands[F, K, V] => List[F[Unit]]): F[Unit]
  def transact[A](ops: RedisCommands[F, K, V] => TxStore[F, String, A] => List[F[Unit]]): F[Map[String, A]]
}

object Transactor {
  private type Cmd[F[_], K, V] = RedisCommands[F, K, V]

  def make[F[_], K, V](redisR: Resource[F, Cmd[F, K, V]], nTransactionPools: Int)(implicit
      F: Async[F]
  ): Resource[F, Transactor[F, K, V]] = {
    assert(nTransactionPools >= 1, "nTransactionPools must be positive")
    for {
      cmds   <- List.fill(nTransactionPools)(redisR).sequence
      queue <- Resource.eval(Queue.bounded[F, Cmd[F, K, V]](nTransactionPools))
      _          <- Resource.eval(cmds.traverse(queue.offer))
    } yield new Transactor[F, K, V] {

      override def transact_(ops: Cmd[F, K, V] => List[F[Unit]]): F[Unit] =
        queue.take.flatMap { cmd =>
          cmd.transact_(ops(cmd)).guarantee(queue.offer(cmd).uncancelable)
        }

      override def transact[A](ops: Cmd[F, K, V] => TxStore[F, String, A] => List[F[Unit]]): F[Map[String, A]] =
        queue.take.flatMap { cmd =>
          cmd.transact(ops(cmd)).guarantee(queue.offer(cmd).uncancelable)
        }
    }
  }
}

Example usage (based on the one in the docs):

object Main extends IOApp.Simple {

  val key1 = "test1"
  val key2 = "test2"
  val key3 = "test3"
  val showResult: String => Option[String] => IO[Unit] = key =>
    _.fold(Log[IO].info(s"Key not found: $key"))(s => Log[IO].info(s"$key: $s"))

  override def run: IO[Unit] = {
    val redisR      = Redis[IO].utf8("redis://localhost")
    val transactorR = Transactor.make(redisR, 2)

    Resource.both(redisR, transactorR).use { case (redis, tx) =>
      val setters = redis.set(key2, "delete_me") >> redis.set(key3, "foo")

      val getters =
        redis.get(key1).flatTap(showResult(key1)) >>
          redis.get(key2).flatTap(showResult(key2))

      val ops = (cmd: RedisCommands[IO, String, String]) =>
        (store: TxStore[IO, String, Option[String]]) =>
          List(
            cmd.set(key1, "foo"),
            cmd.del(key2).void,
            cmd.get(key3).flatMap(store.set(key3))
          )

      val prog =
        tx
          .transact(ops(_))
          .flatMap { kv => IO.println(s"KV: ${kv}") }
          .recoverWith {
            case TransactionDiscarded => Log[IO].error("[Error] - Transaction Discarded")
            case e                    => Log[IO].error(s"[Error] - $e")
          }
      setters >> getters >> prog >> getters.void
    }
  }
}