Merge pull request #16 from Pritesh-Patel/master

Redis sentinel support

README.md

@@ -3,13 +3,13 @@
 ## What is Akka Persistence Redis plugin?
 This is a plugin for Akka Persistence that uses Redis as backend. It uses [rediscala](https://github.com/etaty/rediscala), an asynchronous Redis client written with Akka and Scala.
 It also depends on play-json for JSON serialization.
- 
+
 ## Compatibility
 ### Akka 2.4.x and Play 2.5.x
 Use versions from *0.4.0* for Akka 2.4.x and Play 2.5.x
 ```
 resolvers += Resolver.jcenterRepo // Adds Bintray to resolvers for akka-persistence-redis and rediscala
-libraryDependencies ++= Seq("com.hootsuite" %% "akka-persistence-redis" % "0.4.0")
+libraryDependencies ++= Seq("com.hootsuite" %% "akka-persistence-redis" % "0.5.0")
 ```
 
 ### Akka 2.4.x and Play 2.4.x
@@ -27,9 +27,9 @@ libraryDependencies ++= Seq("com.hootsuite" %% "akka-persistence-redis" % "0.2.2
 ```
 Deprecated methods since Akka 2.3.4 are NOT implemented. As a result, some tests in [TCK](http://doc.akka.io/docs/akka/snapshot/scala/persistence.html#Plugin_TCK) fail.
 Rest of methods are tested with the test harness included in this project.
-  
+
 ### SNAPSHOT
-Development snapshots are published on JFrog OSS 
+Development snapshots are published on JFrog OSS
 `resolvers += "akka-persistence-redis" at "http://oss.jfrog.org/oss-snapshot-local"`
 
 ## Quick start guide
@@ -53,13 +53,22 @@ redis {
 }
 ```
 
+If using sentinel
+```
+redis {
+  sentinel = true
+  sentinel-master = "mymaster"  //master name
+  sentinels = [{host :"localhost", port: 26379}] // list of sentinel addresses
+}
+```
+
 ## Usage
 ### Redis keys
 Journal and snapshot use "journal:" and "snapshot:" as part of keys respectively as it is a good practice to create namespace for keys [reference](https://redislabs.com/blog/5-key-takeaways-for-developing-with-redis)
-  
+
 ### How is data stored in Redis?
 In order to enforce ordering, journal entries and snapshots are inserted into [Sorted Set](http://redis.io/commands#sorted_set) and sorted by sequenceNr
- 
+
 ### Using custom ExecutionContext
 By default, global ExecutionContext is used for Redis operation. This blocks calling thread for synchronous Akka Persistence APIs.
 Override JournalExecutionContext trait to use custom thread pool if blocking in global is undesirable.
@@ -70,7 +79,7 @@ trait CustomExecutionContext extends JournalExecutionContext {
 ```
 
 ## How to contribute
-Contribute by submitting a PR and a bug report in GitHub. 
+Contribute by submitting a PR and a bug report in GitHub.
 
 ## Maintainer
 [Steve Song](https://github.com/ssong-van) [@ssongvan](https://twitter.com/ssongvan)

project/Version.scala

@@ -1,5 +1,5 @@
 object Version {
-  val project = "0.4.0"
+  val project = "0.5.0"
   val scala = "2.11.7"
   val akka = "2.4.4"
   val play = "2.5.2"

src/main/resources/reference.conf

@@ -2,6 +2,9 @@
 redis {
   host = "localhost"
   port = 6379
+  sentinel = false
+  #sentinel-master = "mymaster"
+  #sentinels = [{host :"localhost", port: 26379}]
 }
 
 # This enables akka-persistence-redis plugin

src/main/scala/com/hootsuite/akka/persistence/redis/DefaultRedisComponent.scala

@@ -3,12 +3,19 @@ package com.hootsuite.akka.persistence.redis
 import akka.actor.ActorSystem
 import redis.RedisClient
 
+
 trait DefaultRedisComponent {
   implicit val actorSystem: ActorSystem
 
   private val config = actorSystem.settings.config
-  private val host = config.getString("redis.host")
-  private val port = config.getInt("redis.port")
+  private val sentinel = config.getBoolean("redis.sentinel")
+
+  lazy val redis = if(sentinel){
+    SentinelUtils.getSentinelBasedClient(config)
+  } else {
+    val host = config.getString("redis.host")
+    val port = config.getInt("redis.port")
+    RedisClient(host, port)
+  }
 
-  lazy val redis = new RedisClient(host, port)
 }

src/main/scala/com/hootsuite/akka/persistence/redis/SentinelUtils.scala

@@ -0,0 +1,19 @@
+package com.hootsuite.akka.persistence.redis
+
+import akka.actor.ActorSystem
+import com.typesafe.config.Config
+import redis.{SentinelMonitoredRedisClient, RedisClient}
+import scala.collection.JavaConversions._
+
+object SentinelUtils {
+
+  def getSentinelBasedClient(config: Config)(implicit system:ActorSystem):RedisClient = {
+
+    val sentinelMaster = config.getString("redis.sentinel-master")
+    val sentinels = config.getConfigList("redis.sentinels") map { conf =>
+      conf.getString("host") -> conf.getInt("port")
+    }
+    SentinelMonitoredRedisClient(sentinels,sentinelMaster).redisClient
+  }
+
+}

src/test/resources/reference.conf

@@ -2,4 +2,7 @@
 redis {
   host = "localhost"
   port = 6379
+  sentinel = false
+  #sentinel-master = "mymaster"
+  #sentinels = [{host :"localhost", port: 26379}]
 }

src/test/scala/com/hootsuite/akka/persistence/redis/journal/RedisJournalSpec.scala

@@ -2,6 +2,7 @@ package com.hootsuite.akka.persistence.redis.journal
 
 import akka.persistence.CapabilityFlag
 import akka.persistence.journal.JournalSpec
+import com.hootsuite.akka.persistence.redis.SentinelUtils
 import com.typesafe.config.ConfigFactory
 import redis.RedisClient
 
@@ -21,9 +22,16 @@ class RedisJournalSpec extends JournalSpec(
   override def beforeAll() = {
     super.beforeAll()
     val config = ConfigFactory.load()
-    val host = config.getString("redis.host")
-    val port = config.getInt("redis.port")
-    val redis = new RedisClient(host, port)
+
+    val sentinel = config.getBoolean("redis.sentinel")
+
+    lazy val redis = if(sentinel){
+      SentinelUtils.getSentinelBasedClient(config)
+    } else {
+      val host = config.getString("redis.host")
+      val port = config.getInt("redis.port")
+      new RedisClient(host, port)
+    }
 
     // Clean up database to prevent data from previous tests interfering with current run
     redis.flushall()

src/test/scala/com/hootsuite/akka/persistence/redis/snapshot/RedisSnapshotSpec.scala

@@ -1,6 +1,7 @@
 package com.hootsuite.akka.persistence.redis.snapshot
 
 import akka.persistence.snapshot.SnapshotStoreSpec
+import com.hootsuite.akka.persistence.redis.SentinelUtils
 import com.typesafe.config.ConfigFactory
 import redis.RedisClient
 
@@ -18,9 +19,15 @@ class RedisSnapshotSpec extends SnapshotStoreSpec(
   override def beforeAll() = {
     super.beforeAll()
     val config = ConfigFactory.load()
-    val host = config.getString("redis.host")
-    val port = config.getInt("redis.port")
-    val redis = new RedisClient(host, port)
+    val sentinel = config.getBoolean("redis.sentinel")
+
+    lazy val redis = if(sentinel){
+      SentinelUtils.getSentinelBasedClient(config)
+    } else {
+      val host = config.getString("redis.host")
+      val port = config.getInt("redis.port")
+      new RedisClient(host, port)
+    }
 
     // Clean up database to prevent data from previous tests interfering with current run
     redis.flushall()