RedisCluster -> Mutable/Immutable

src/main/scala/redis/ClusterCommandRedisCluster.scala

@@ -0,0 +1,35 @@
+package redis
+
+import java.net.InetAddress
+import java.time.Duration
+
+import akka.actor.ActorSystem
+
+import scala.concurrent.Future
+
+/**
+  * A mutable Redis cluster that uses the CLUSTER NODES command, which broadcasts out to
+  * all nodes in the cluster and returns a list of nodes, to connect to new nodes and
+  * disconnect from old nodes automatically
+  */
+class ClusterCommandRedisCluster(configEndpoint: String,
+                                 port: Option[Int] = None,
+                                 timeout: Option[Duration] = Some(Duration.ofSeconds(30)))
+                                (implicit _system: ActorSystem) extends MutableRedisCluster(timeout, Seq()) {
+
+  val ipPortPattern = """(([0-9]{1,3}\.){3}[0-9]{1,3})\:([0-9]+).*"""r("ip", "unused", "port")
+
+  def getNodesFromCluster(): Future[Seq[RedisServer]] = {
+    clusterNodes().map(_.map(info => {
+      val ipPortPattern(ip, _, port) = info.ip_port
+      RedisServer(ip, port.toInt)
+    }))
+  }
+
+  private def cnameToARecords(cname: String): Seq[String] = InetAddress.getAllByName(cname).map(_.getHostAddress)
+
+  private def seedCluster(): Unit = cnameToARecords(configEndpoint).map(RedisServer(_, port.getOrElse(6379))).foreach(addServer)
+
+  seedCluster()
+  asyncRefreshClusterSlots(true)
+}

src/main/scala/redis/ImmutableRedisCluster.scala

@@ -0,0 +1,28 @@
+package redis
+
+import java.util.concurrent.TimeUnit
+
+import akka.actor.ActorSystem
+
+import scala.concurrent.Await
+import scala.concurrent.duration.Duration
+
+
+/*
+ * An immutable Redis cluster implementation. Only the servers that are passed to the constructor will be used.
+ */
+class ImmutableRedisCluster(val redisServers: Seq[RedisServer],
+                            override val name: String = "RedisClientPool",
+                            password: Option[String] = None)
+                           (implicit _system: ActorSystem,
+                            redisDispatcher: RedisDispatcher = RedisDispatcher("rediscala.rediscala-client-worker-dispatcher")
+                           ) extends RedisCluster {
+  override val redisServerConnections = {
+    redisServers.map { server =>
+      makeRedisConnection(server, defaultActive = true)
+    } toMap
+  }
+
+  refreshConnections()
+  Await.result(asyncRefreshClusterSlots(force=true), Duration(10,TimeUnit.SECONDS))
+}

src/main/scala/redis/MutableRedisCluster.scala

@@ -0,0 +1,105 @@
+package redis
+
+import akka.actor.ActorSystem
+import java.time.Duration
+
+import scala.concurrent.Future
+import scala.concurrent.duration._
+import scala.concurrent.stm.Ref
+
+
+/*
+ * A Redis cluster that can dynamically add and remove nodes.
+ * Refreshes the node listing automatically based on a timeout parameter.
+ */
+abstract class MutableRedisCluster(timeout: Option[Duration],
+                          val initialServers: Seq[RedisServer],
+                          override val name: String = "RedisClientPool",
+                          password: Option[String] = None)
+                         (implicit _system: ActorSystem,
+                          redisDispatcher: RedisDispatcher = RedisDispatcher("rediscala.rediscala-client-worker-dispatcher")
+                         ) extends RedisCluster {
+
+  def refreshNodeReferences(): Future[Unit] = refreshNodeReferencesHelper(false)
+
+  def getNodesFromCluster(): Future[Seq[RedisServer]]
+
+  override val redisServerConnections: collection.mutable.Map[RedisServer, RedisConnection] = {
+    collection.mutable.Map() ++ initialServers.map(makeConnection).toMap[RedisServer, RedisConnection]
+  }
+
+  private def makeConnection(server: RedisServer): (RedisServer, RedisConnection) =
+    makeRedisConnection(
+      server = server.copy(password = password, db = None),
+      defaultActive = true
+    )
+
+  def redisServers: Seq[RedisServer] = redisServerConnections.synchronized {
+    redisServerConnections.keys.toSeq
+  }
+
+  def addServer(server: RedisServer) {
+    if (!redisServerConnections.contains(server)) {
+      redisServerConnections.synchronized {
+        if (!redisServerConnections.contains(server)) {
+          log.info(s"Adding connection: $server")
+          redisServerConnections += makeRedisConnection(server)
+        }
+      }
+    }
+  }
+
+  def removeServer(askServer: RedisServer) {
+    if (redisServerConnections.contains(askServer)) {
+      log.info(s"Removing connection: $askServer")
+      redisServerConnections.synchronized {
+        redisServerConnections.get(askServer).foreach { redisServerConnection =>
+          _system stop redisServerConnection.actor
+        }
+        redisServerConnections.remove(askServer)
+        refreshConnections()
+      }
+    }
+  }
+
+  protected val lockRefreshNodes = Ref(false)
+
+  protected def refreshNodeReferencesHelper(force: Boolean = false): Future[Unit] = {
+    if (lockRefreshNodes.single.compareAndSet(false, true)) {
+      for {
+        refreshedNodes <- getNodesFromCluster()
+      } yield {
+        val currentNodes = redisServerConnections.toSeq.map(_._1)
+        log.debug(s"refreshNodeReferences: $refreshedNodes")
+        refreshedNodes match {
+          case Nil => log.warning("Refreshed nodes is an empty list - not updating connections")
+          case _ =>
+            refreshedNodes.foreach(node => {
+              if (!currentNodes.contains(node)) {
+                addServer(node)
+              }
+            })
+            currentNodes.foreach(node => {
+              if (!refreshedNodes.contains(node)) {
+                removeServer(node)
+              }
+            })
+        }
+        refreshConnections()
+        asyncRefreshClusterSlots(force)
+      }
+      lockRefreshNodes.single.compareAndSet(true, false)
+    }
+    Future.successful(())
+  }
+
+  protected val timeoutMillis: Option[Long] = timeout.map(_.toMillis)
+
+  if(timeout.isDefined) {
+    val loopingTimer = timeout.get.getSeconds.seconds
+    log.info(s"Registering elasticache node refresh on a loop of $loopingTimer seconds.")
+    _system.scheduler.schedule(initialDelay = loopingTimer, interval = loopingTimer)({
+      refreshNodeReferences()
+    })
+  }
+}

src/main/scala/redis/RedisCluster.scala

@@ -1,6 +1,6 @@
 package redis
 
-import java.util.concurrent.{ThreadLocalRandom, TimeUnit}
+import java.util.concurrent.ThreadLocalRandom
 
 import akka.actor.{ActorRef, ActorSystem}
 import akka.event.Logging
@@ -9,32 +9,21 @@ import redis.api.clusters.{ClusterNode, ClusterSlot}
 import redis.protocol.RedisReply
 import redis.util.CRC16
 
-import scala.concurrent.duration.Duration
 import scala.concurrent.stm.Ref
-import scala.concurrent.{Await, Future, Promise}
+import scala.concurrent.{Future, Promise}
 import scala.util.control.NonFatal
 
 
-case class RedisCluster(redisServers: Seq[RedisServer],
-                           name: String = "RedisClientPool")
-                          (implicit _system: ActorSystem,
-                           redisDispatcher: RedisDispatcher = Redis.dispatcher
-                          ) extends RedisClientPoolLike(_system, redisDispatcher)  with RedisCommands {
+abstract class RedisCluster(override val name: String = "RedisClientPool")
+                           (implicit _system: ActorSystem,
+                            redisDispatcher: RedisDispatcher = Redis.dispatcher
+                           ) extends RedisClientPoolLike(_system, redisDispatcher) with RedisCommands {
 
-  val log = Logging.getLogger(_system, this)
+  def redisServers: Seq[RedisServer]
 
-  val clusterSlotsRef:Ref[Option[Map[ClusterSlot, RedisConnection]]] = Ref(Option.empty[Map[ClusterSlot, RedisConnection]])
-  val lockClusterSlots = Ref(true)
+  protected val log = Logging.getLogger(_system, this)
 
-  override val redisServerConnections = {
-    redisServers.map { server =>
-      makeRedisConnection(server, defaultActive = true)
-    } toMap
-  }
-  refreshConnections()
-
-
-  def equalsHostPort(clusterNode:ClusterNode,server:RedisServer) = {
+  protected def equalsHostPort(clusterNode:ClusterNode,server:RedisServer) = {
     clusterNode.host == server.host &&  clusterNode.port == server.port
   }
 
@@ -43,7 +32,7 @@ case class RedisCluster(redisServers: Seq[RedisServer],
       if (active.single.compareAndSet(!status, status)) {
         refreshConnections()
       }
-      
+
       clusterSlotsRef.single.get.map { clusterSlots =>
         if (clusterSlots.keys.exists( cs => equalsHostPort(cs.master,server) )){
           log.info("one master is still dead => refresh clusterSlots")
@@ -54,7 +43,10 @@ case class RedisCluster(redisServers: Seq[RedisServer],
     }
   }
 
-  def getClusterSlots(): Future[Map[ClusterSlot, RedisConnection]] = {
+  protected val clusterSlotsRef: Ref[Option[Map[ClusterSlot, RedisConnection]]] = Ref(Option.empty[Map[ClusterSlot, RedisConnection]])
+  protected val lockClusterSlots = Ref(true)
+
+  protected def getClusterSlots(): Future[Map[ClusterSlot, RedisConnection]] = {
 
     def resolveClusterSlots(retry:Int): Future[Map[ClusterSlot, RedisConnection]] = {
       clusterSlots().map { clusterSlots =>
@@ -74,28 +66,33 @@ case class RedisCluster(redisServers: Seq[RedisServer],
     resolveClusterSlots(3) //retry 3 times
   }
 
-  def asyncRefreshClusterSlots(force:Boolean=false): Future[Unit] = {
-    if( force || lockClusterSlots.single.compareAndSet(false,true) ) {
-     try {
-       getClusterSlots().map { clusterSlot =>
-         log.info("refreshClusterSlots: " + clusterSlot.toString())
-         clusterSlotsRef.single.set(Some(clusterSlot))
-         lockClusterSlots.single.compareAndSet(true, false)
-         ()
-       }.recoverWith {
-         case NonFatal(e) =>
-           log.error("refreshClusterSlots:",e)
-           lockClusterSlots.single.compareAndSet(true, false)
-           Future.failed(e)
-       }
-     }catch{
-       case NonFatal(e) =>
-         lockClusterSlots.single.compareAndSet(true, false)
-         throw e
-     }
-    }else{
-
-     Future.successful(clusterSlotsRef.single.get)
+  protected def asyncRefreshClusterSlots(force: Boolean = false, retry: Int = 3): Future[Unit] = {
+    if (force || lockClusterSlots.single.compareAndSet(false, true)) {
+      try {
+        getClusterSlots().map { clusterSlot =>
+          log.info("refreshClusterSlots: " + clusterSlot.toString())
+          clusterSlotsRef.single.set(Some(clusterSlot))
+          lockClusterSlots.single.compareAndSet(true, false)
+          ()
+        }.recoverWith {
+          case NonFatal(e) =>
+            log.warning("refreshClusterSlots:", e)
+            lockClusterSlots.single.compareAndSet(true, false)
+            if(retry > 0){
+              log.warning("refreshClusterSlots: Retrying...")
+              Thread.sleep(250L)
+              asyncRefreshClusterSlots(force, retry-1)
+            } else {
+              Future.failed(e)
+            }
+        }
+      } catch {
+        case NonFatal(e) =>
+          lockClusterSlots.single.compareAndSet(true, false)
+          throw e
+      }
+    } else {
+      Future.successful(clusterSlotsRef.single.get)
     }
   }
 
@@ -105,7 +102,7 @@ case class RedisCluster(redisServers: Seq[RedisServer],
     promise.future
   }
 
-  def getRedisConnection(slot:Int):Option[RedisConnection] = {
+  protected def getRedisConnection(slot:Int):Option[RedisConnection] = {
         getClusterAndConnection(slot)
           .map{ case ( _,redisConnection ) => redisConnection  }
 
@@ -124,7 +121,7 @@ case class RedisCluster(redisServers: Seq[RedisServer],
     }
   }
 
-  val redirectMessagePattern = """(MOVED|ASK) \d+ (\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}):(\d+)""".r
+  protected val redirectMessagePattern = """(MOVED|ASK) \d+ (\d{1,3}\.\d{1,3}\.\d{1,3}\.\d{1,3}):(\d+)""".r
   override def send[T](redisCommand: RedisCommand[_ <: RedisReply, T]): Future[T] = {
 
     val maybeRedisActor:Option[ActorRef]  = getRedisActor(redisCommand)
@@ -158,7 +155,7 @@ case class RedisCluster(redisServers: Seq[RedisServer],
     }.getOrElse(Future.failed(new RuntimeException("server not found: no server available")))
   }
 
-  def getRedisActor[T](redisCommand: RedisCommand[_ <: RedisReply, T]): Option[ActorRef] = {
+  protected def getRedisActor[T](redisCommand: RedisCommand[_ <: RedisReply, T]): Option[ActorRef] = {
     redisCommand match {
       case clusterKey: ClusterKey =>
         getRedisConnection(clusterKey.getSlot())
@@ -176,13 +173,11 @@ case class RedisCluster(redisServers: Seq[RedisServer],
     }
   }
 
-  def groupByCluserServer(keys:Seq[String]): Seq[Seq[String]] = {
+  def groupByClusterServer(keys:Seq[String]): Seq[Seq[String]] = {
     keys.groupBy{
       key => getRedisConnection(RedisComputeSlot.hashSlot(key))
     }.values.toSeq
   }
-
-  Await.result(asyncRefreshClusterSlots(force=true), Duration(10,TimeUnit.SECONDS))
 }