GitHub - etaty/rediscala: Non-blocking, Reactive Redis driver for Scala (with Sentinel support)

Non-blocking, Reactive Redis driver for Scala (with Sentinel support) - etaty/rediscala

Visit Site

GitHub - etaty/rediscala: Non-blocking, Reactive Redis driver for Scala (with Sentinel support)

Non-blocking, Reactive Redis driver for Scala (with Sentinel support) - etaty/rediscala

rediscala

A Redis client for Scala with non-blocking and asynchronous I/O operations.

Reactive : Redis requests/replies are wrapped in Futures.
Typesafe : Redis types are mapped to Scala types.
Fast : Rediscala uses redis pipelining. Blocking redis commands are moved into their own connection. A worker actor handles I/O operations (I/O bounds), another handles decoding of Redis replies (CPU bounds).

Set up your project dependencies

If you use SBT, you just have to edit build.sbt and add the following:

From version 1.9.0:

use akka 2.5.23 (java 1.8)
released for scala
- 2.11
- 2.12
- 2.13

libraryDependencies += "com.github.etaty" %% "rediscala" % "1.9.0"

From version 1.8.0:

use akka 2.4.12 (java 1.8)
released for scala 2.11 & 2.12

libraryDependencies += "com.github.etaty" %% "rediscala" % "1.8.0"

From version 1.3.1:

use akka 2.3
released for scala 2.10 & 2.11

// new repo on maven.org
libraryDependencies += "com.github.etaty" %% "rediscala" % "1.7.0"


// old repo on bintray (1.5.0 and inferior version)
resolvers += "rediscala" at "http://dl.bintray.com/etaty/maven"
libraryDependencies += "com.etaty.rediscala" %% "rediscala" % "1.5.0"

For older rediscala versions (<= 1.3):

use akka 2.2
released for scala 2.10 only
use github "repo"

resolvers += "rediscala" at "https://raw.github.com/etaty/rediscala-mvn/master/releases/"

libraryDependencies += "com.etaty.rediscala" %% "rediscala" % "1.3"

Connect to the database

import redis.RedisClient
import scala.concurrent.Await
import scala.concurrent.duration._
import scala.concurrent.ExecutionContext.Implicits.global

object Main extends App {
  implicit val akkaSystem = akka.actor.ActorSystem()

  val redis = RedisClient()

  val futurePong = redis.ping()
  println("Ping sent!")
  futurePong.map(pong => {
    println(s"Redis replied with a $pong")
  })
  Await.result(futurePong, 5 seconds)

  akkaSystem.shutdown()
}

Basic Example

https://github.com/etaty/rediscala-demo

You can fork with : git clone [email protected]:etaty/rediscala-demo.git then run it, with sbt run

Redis Commands

All commands are supported :

Keys (scaladoc)
Strings (scaladoc)
Hashes (scaladoc)
Lists
- non-blocking (scaladoc)
- blocking (scaladoc)
Sets (scaladoc)
Sorted Sets (scaladoc)
Pub/Sub (scaladoc)
Transactions (scaladoc)
Connection (scaladoc)
Scripting (scaladoc)
Server (scaladoc)
HyperLogLog (scaladoc)

Blocking commands

RedisBlockingClient is the instance allowing access to blocking commands :

blpop
brpop
brpopplush

  redisBlocking.blpop(Seq("workList", "otherKeyWithWork"), 5 seconds).map(result => {
    result.map({
      case (key, work) => println(s"list $key has work : ${work.utf8String}")
    })
  })

Full example: ExampleRediscalaBlocking

You can fork with: git clone [email protected]:etaty/rediscala-demo.git then run it, with sbt run

Transactions

The idea behind transactions in Rediscala is to start a transaction outside of a redis connection. We use the TransactionBuilder to store call to redis commands (and for each command we give back a future). When exec is called, TransactionBuilder will build and send all the commands together to the server. Then the futures will be completed. By doing that we can use a normal connection with pipelining, and avoiding to trap a command from outside, in the transaction...

  val redisTransaction = redis.transaction() // new TransactionBuilder
  redisTransaction.watch("key")
  val set = redisTransaction.set("key", "abcValue")
  val decr = redisTransaction.decr("key")
  val get = redisTransaction.get("key")
  redisTransaction.exec()

Full example: ExampleTransaction

You can fork with : git clone [email protected]:etaty/rediscala-demo.git then run it, with sbt run

TransactionsSpec will reveal even more gems of the API.

Pub/Sub

You can use a case class with callbacks RedisPubSub or extend the actor RedisSubscriberActor as shown in the example below

object ExamplePubSub extends App {
  implicit val akkaSystem = akka.actor.ActorSystem()

  val redis = RedisClient()

  // publish after 2 seconds every 2 or 5 seconds
  akkaSystem.scheduler.schedule(2 seconds, 2 seconds)(redis.publish("time", System.currentTimeMillis()))
  akkaSystem.scheduler.schedule(2 seconds, 5 seconds)(redis.publish("pattern.match", "pattern value"))
  // shutdown Akka in 20 seconds
  akkaSystem.scheduler.scheduleOnce(20 seconds)(akkaSystem.shutdown())

  val channels = Seq("time")
  val patterns = Seq("pattern.*")
  // create SubscribeActor instance
  akkaSystem.actorOf(Props(classOf[SubscribeActor], channels, patterns).withDispatcher("rediscala.rediscala-client-worker-dispatcher"))

}

class SubscribeActor(channels: Seq[String] = Nil, patterns: Seq[String] = Nil) extends RedisSubscriberActor(channels, patterns) {
  override val address: InetSocketAddress = new InetSocketAddress("localhost", 6379)

  def onMessage(message: Message) {
    println(s"message received: $message")
  }

  def onPMessage(pmessage: PMessage) {
    println(s"pattern message received: $pmessage")
  }
}

Full example: ExamplePubSub

You can fork with : git clone [email protected]:etaty/rediscala-demo.git then run it, with sbt run

RedisPubSubSpec will reveal even more gems of the API.

Scripting

RedisScript is a helper, you can put your LUA script inside and it will compute the hash. You can use it with evalshaOrEval which run your script even if it wasn't already loaded.

  val redis = RedisClient()

  val redisScript = RedisScript("return 'rediscala'")

  val r = redis.evalshaOrEval(redisScript).map({
    case b: Bulk => println(b.toString())
  })
  Await.result(r, 5 seconds)

Full example: ExampleScripting

Redis Sentinel

SentinelClient connect to a redis sentinel server.

SentinelMonitoredRedisClient connect to a sentinel server to find the master addresse then start a connection. In case the master change your RedisClient connection will automatically connect to the new master server. If you are using a blocking client, you can use SentinelMonitoredRedisBlockingClient

Pool

RedisClientPool connect to a pool of redis servers. Redis commands are dispatched to redis connection in a round robin way.

Master Slave

RedisClientMasterSlaves connect to a master and a pool of slaves. The write commands are sent to the master, while the read commands are sent to the slaves in the RedisClientPool

Config Which Dispatcher to Use

By default, the actors in this project will use the dispatcher rediscala.rediscala-client-worker-dispatcher. If you want to use another dispatcher, just config the implicit value of redisDispatcher:

implicit val redisDispatcher = RedisDispatcher("akka.actor.default-dispatcher")

ByteStringSerializer ByteStringDeserializer ByteStringFormatter

ByteStringSerializer

ByteStringDeserializer

ByteStringFormatter

case class DumbClass(s1: String, s2: String)

object DumbClass {
  implicit val byteStringFormatter = new ByteStringFormatter[DumbClass] {
    def serialize(data: DumbClass): ByteString = {
      //...
    }

    def deserialize(bs: ByteString): DumbClass = {
      //...
    }
  }
}
//...

  val dumb = DumbClass("s1", "s2")

  val r = for {
    set <- redis.set("dumbKey", dumb)
    getDumbOpt <- redis.get[DumbClass]("dumbKey")
  } yield {
    getDumbOpt.map(getDumb => {
      assert(getDumb == dumb)
      println(getDumb)
    })
  }

Full example: ExampleByteStringFormatter