-
Notifications
You must be signed in to change notification settings - Fork 12
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #3 from ChristopherDavenport/transactions
- Loading branch information
Showing
8 changed files
with
547 additions
and
354 deletions.
There are no files selected for viewing
642 changes: 320 additions & 322 deletions
642
core/src/main/scala/io/chrisdavenport/rediculous/RedisCommands.scala
Large diffs are not rendered by default.
Oops, something went wrong.
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
17 changes: 17 additions & 0 deletions
17
core/src/main/scala/io/chrisdavenport/rediculous/RedisCtx.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,17 @@ | ||
| package io.chrisdavenport.rediculous | ||
|
|
||
| import cats.data.NonEmptyList | ||
| import cats.effect.Concurrent | ||
|
|
||
| trait RedisCtx[F[_]]{ | ||
| def run[A: RedisResult](command: NonEmptyList[String]): F[A] | ||
| } | ||
|
|
||
| object RedisCtx { | ||
| def apply[F[_]](implicit ev: RedisCtx[F]): ev.type = ev | ||
|
|
||
| implicit def redis[F[_]: Concurrent]: RedisCtx[Redis[F, *]] = new RedisCtx[Redis[F, *]]{ | ||
| def run[A: RedisResult](command: NonEmptyList[String]): Redis[F, A] = | ||
| RedisConnection.runRequestTotal(command) | ||
| } | ||
| } |
125 changes: 125 additions & 0 deletions
125
core/src/main/scala/io/chrisdavenport/rediculous/RedisTransaction.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,125 @@ | ||
| package io.chrisdavenport.rediculous | ||
|
|
||
| import cats._ | ||
| import cats.implicits._ | ||
| import cats.data._ | ||
| import cats.effect._ | ||
| import RedisProtocol._ | ||
|
|
||
|
|
||
| /** | ||
| * Transactions Operate via typeclasses. RedisCtx allows us to abstract our operations into | ||
| * different types depending on the behavior we want. In the case of transactions that is | ||
| * [[RedisTransaction]]. These can be composed together via its Applicative | ||
| * instance to form a transaction consisting of multiple commands, then transacted via | ||
| * either multiExec or transact on the class. | ||
| * | ||
| * @example | ||
| * {{{ | ||
| * import io.chrisdavenport.rediculous._ | ||
| * import cats.effect.Concurrent | ||
| * val tx = ( | ||
| * RedisCommands.ping[RedisTransaction], | ||
| * RedisCommands.del[RedisTransaction](List("foo")), | ||
| * RedisCommands.get[RedisTransaction]("foo"), | ||
| * RedisCommands.set[RedisTransaction]("foo", "value"), | ||
| * RedisCommands.get[RedisTransaction]("foo") | ||
| * ).tupled | ||
| * | ||
| * def operation[F[_]: Concurrent] = tx.transact[F] | ||
| * }}} | ||
| **/ | ||
| final case class RedisTransaction[A](value: RedisTransaction.RedisTxState[RedisTransaction.Queued[A]]){ | ||
| def transact[F[_]: Concurrent]: Redis[F, RedisTransaction.TxResult[A]] = | ||
| RedisTransaction.multiExec[F](this) | ||
| } | ||
|
|
||
| object RedisTransaction { | ||
|
|
||
| implicit val ctx: RedisCtx[RedisTransaction] = new RedisCtx[RedisTransaction]{ | ||
| def run[A: RedisResult](command: NonEmptyList[String]): RedisTransaction[A] = RedisTransaction(RedisTxState{for { | ||
| (i, base) <- State.get | ||
| _ <- State.set((i + 1, command :: base)) | ||
| } yield Queued(l => RedisResult[A].decode(l(i)))}) | ||
| } | ||
| implicit val applicative: Applicative[RedisTransaction] = new Applicative[RedisTransaction]{ | ||
| def pure[A](a: A) = RedisTransaction(Monad[RedisTxState].pure(Monad[Queued].pure(a))) | ||
|
|
||
| override def ap[A, B](ff: RedisTransaction[A => B])(fa: RedisTransaction[A]): RedisTransaction[B] = | ||
| RedisTransaction(RedisTxState( | ||
| Nested(ff.value.value).ap(Nested(fa.value.value)).value | ||
| )) | ||
| } | ||
|
|
||
| /** | ||
| * A TxResult Represent the state of a RedisTransaction when run. | ||
| * Success means it completed succesfully, Aborted means we received | ||
| * a Nil Arrary from Redis which represent that at least one key being watched | ||
| * has been modified. An error occurs depending on the succesful execution of | ||
| * the function built in Queued. | ||
| */ | ||
| sealed trait TxResult[+A] | ||
| object TxResult { | ||
| final case class Success[A](value: A) extends TxResult[A] | ||
| final case object Aborted extends TxResult[Nothing] | ||
| final case class Error(value: String) extends TxResult[Nothing] | ||
| } | ||
|
|
||
| final case class RedisTxState[A](value: State[(Int, List[NonEmptyList[String]]), A]) | ||
| object RedisTxState { | ||
|
|
||
| implicit val m: Monad[RedisTxState] = new StackSafeMonad[RedisTxState]{ | ||
| def pure[A](a: A): RedisTxState[A] = RedisTxState(Monad[State[(Int, List[NonEmptyList[String]]), *]].pure(a)) | ||
| def flatMap[A, B](fa: RedisTxState[A])(f: A => RedisTxState[B]): RedisTxState[B] = RedisTxState( | ||
| fa.value.flatMap(f.andThen(_.value)) | ||
| ) | ||
| } | ||
| } | ||
| final case class Queued[A](f: List[Resp] => Either[Resp, A]) | ||
| object Queued { | ||
| implicit val m: Monad[Queued] = new StackSafeMonad[Queued]{ | ||
| def pure[A](a: A) = Queued{_ => Either.right(a)} | ||
| def flatMap[A, B](fa: Queued[A])(f: A => Queued[B]): Queued[B] = { | ||
| Queued{l => | ||
| for { | ||
| a <- fa.f(l) | ||
| b <- f(a).f(l) | ||
| } yield b | ||
| } | ||
| } | ||
| } | ||
| } | ||
|
|
||
| // ---------- | ||
| // Operations | ||
| // ---------- | ||
| def watch[F[_]: Concurrent](keys: List[String]): Redis[F, Status] = | ||
| RedisCtx[Redis[F,*]].run(NonEmptyList("WATCH", keys)) | ||
|
|
||
| def unwatch[F[_]: Concurrent]: Redis[F, Status] = | ||
| RedisCtx[Redis[F,*]].run(NonEmptyList.of("UNWATCH")) | ||
|
|
||
| def multiExec[F[_]] = new MultiExecPartiallyApplied[F] | ||
|
|
||
| class MultiExecPartiallyApplied[F[_]]{ | ||
|
|
||
| def apply[A](tx: RedisTransaction[A])(implicit F: Concurrent[F]): Redis[F, TxResult[A]] = { | ||
| Redis(Kleisli{c: RedisConnection[F] => | ||
| val ((_, commandsR), Queued(f)) = tx.value.value.run((0, List.empty)).value | ||
| val commands = commandsR.reverse | ||
| RedisConnection.runRequestInternal(c)(NonEmptyList( | ||
| NonEmptyList.of("MULTI"), | ||
| commands ++ | ||
| List(NonEmptyList.of("EXEC")) | ||
| )).map{_.flatMap{_.last match { | ||
| case Resp.Array(Some(a)) => f(a).fold[TxResult[A]](e => TxResult.Error(e.toString), TxResult.Success(_)).pure[F] | ||
| case Resp.Array(None) => (TxResult.Aborted: TxResult[A]).pure[F] | ||
| case other => ApplicativeError[F, Throwable].raiseError(new Throwable(s"EXEC returned $other")) | ||
| }}} | ||
| }) | ||
| } | ||
| } | ||
|
|
||
|
|
||
|
|
||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,36 @@ | ||
| import io.chrisdavenport.rediculous._ | ||
| import cats.implicits._ | ||
| import cats.effect._ | ||
| import fs2.io.tcp._ | ||
| import java.net.InetSocketAddress | ||
|
|
||
| // Send a Single Transaction to the Redis Server | ||
| object TransactionExample extends IOApp { | ||
|
|
||
| def run(args: List[String]): IO[ExitCode] = { | ||
| val r = for { | ||
| blocker <- Blocker[IO] | ||
| sg <- SocketGroup[IO](blocker) | ||
| // maxQueued: How many elements before new submissions semantically block. Tradeoff of memory to queue jobs. | ||
| // Default 1000 is good for small servers. But can easily take 100,000. | ||
| // workers: How many threads will process pipelined messages. | ||
| connection <- RedisConnection.queued[IO](sg, new InetSocketAddress("localhost", 6379), maxQueued = 10000, workers = 2) | ||
| } yield connection | ||
|
|
||
| r.use {client => | ||
| val r = ( | ||
| RedisCommands.ping[RedisTransaction], | ||
| RedisCommands.del[RedisTransaction](List("foo")), | ||
| RedisCommands.get[RedisTransaction]("foo"), | ||
| RedisCommands.set[RedisTransaction]("foo", "value"), | ||
| RedisCommands.get[RedisTransaction]("foo") | ||
| ).tupled | ||
|
|
||
| val multi = r.transact[IO] | ||
|
|
||
| multi.run(client).flatTap(output => IO(println(output))) | ||
|
|
||
| }.as(ExitCode.Success) | ||
|
|
||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters