Create MappedAllocationStrategy - post-code-review changes

evolution-gaming · Aug 15, 2017 · a925dca · a925dca
1 parent fdb1215
commit a925dca
Show file tree

Hide file tree

Showing 5 changed files with 24 additions and 19 deletions.
diff --git a/build.sbt b/build.sbt
@@ -33,7 +33,7 @@ scalacOptions ++= Seq(
 
 scalacOptions in (Compile,doc) ++= Seq("-no-link-warnings")
 
-val AkkaVersion = "2.5.3"
+val AkkaVersion = "2.5.4"
 
 resolvers += Resolver.bintrayRepo("evolutiongaming", "maven")
 

diff --git a/src/main/scala/com/evolutiongaming/cluster/AdaptiveAllocationStrategyDDProxy.scala b/src/main/scala/com/evolutiongaming/cluster/AdaptiveAllocationStrategyDDProxy.scala
@@ -28,9 +28,9 @@ class AdaptiveAllocationStrategyDDProxy extends Actor with ActorLogging {
   import AdaptiveAllocationStrategyDDProxy._
 
   implicit lazy val node = Cluster(context.system)
-  private val selfAddress = node.selfAddress.toString
+  lazy val selfAddress = node.selfAddress.toString
   lazy val replicator: ActorRef = DistributedData(context.system).replicator
-  private val emptyMap = ORMultiMap.empty[String, String]
+  val emptyMap = ORMultiMap.empty[String, String]
 
   replicator ! Subscribe(EntityToNodeCountersKey, self)
 

diff --git a/src/main/scala/com/evolutiongaming/cluster/ExtendedShardAllocationStrategy.scala b/src/main/scala/com/evolutiongaming/cluster/ExtendedShardAllocationStrategy.scala
@@ -38,22 +38,26 @@ abstract class ExtendedShardAllocationStrategy(
   final def allocateShard(
     requester: ActorRef,
     shardId: ShardRegion.ShardId,
-    currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardRegion.ShardId]]): Future[ActorRef] =
-  for (nodeByStrategy <- doAllocate(requester, shardId, currentShardAllocations)) yield {
+    currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardRegion.ShardId]]): Future[ActorRef] = {
+
+    val nodeByStrategyFuture = doAllocate(requester, shardId, currentShardAllocations)
     val ignoredNodes = nodesToDeallocate()
 
     if (ignoredNodes.isEmpty)
-      nodeByStrategy
+      nodeByStrategyFuture
     else {
-      val activeNodes = notIgnoredNodes(currentShardAllocations, ignoredNodes)
-      val activeAddresses = activeNodes map (_.path.address)
-
-      if (activeAddresses contains (addressHelper toGlobal nodeByStrategy.path.address))
-        nodeByStrategy
-      else if (activeAddresses contains (addressHelper toGlobal requester.path.address))
-        requester
-      else
-        activeNodes.headOption getOrElse currentShardAllocations.keys.head
+      for (nodeByStrategy <- nodeByStrategyFuture) yield {
+
+        val activeNodes = notIgnoredNodes(currentShardAllocations, ignoredNodes)
+        val activeAddresses = activeNodes map (_.path.address)
+
+        if (activeAddresses contains (addressHelper toGlobal nodeByStrategy.path.address))
+          nodeByStrategy
+        else if (activeAddresses contains (addressHelper toGlobal requester.path.address))
+          requester
+        else
+          activeNodes.headOption getOrElse currentShardAllocations.keys.head
+      }
     }
   }
 

diff --git a/src/main/scala/com/evolutiongaming/cluster/MappedAllocationStrategy.scala b/src/main/scala/com/evolutiongaming/cluster/MappedAllocationStrategy.scala
@@ -38,18 +38,19 @@ class MappedAllocationStrategy(
     }
   }
 
-
   def rebalance(
     currentShardAllocations: Map[ActorRef, immutable.IndexedSeq[ShardRegion.ShardId]],
     rebalanceInProgress: Set[ShardRegion.ShardId]): Future[Set[ShardRegion.ShardId]] = {
 
     logger debug
       s"rebalance $typeName: currentShardAllocations = $currentShardAllocations, rebalanceInProgress = $rebalanceInProgress"
 
-    val result = ((for {
+    val shardsToRebalance = for {
       (ref, shards) <- currentShardAllocations
       shardId <- shards if !(shardToRegionMapping get EntityKey(typeName, shardId) contains ref)
-    } yield shardId).toSet -- rebalanceInProgress) take maxSimultaneousRebalance
+    } yield shardId
+
+    val result = (shardsToRebalance.toSet -- rebalanceInProgress) take maxSimultaneousRebalance
 
     if (result.nonEmpty) logger info s"Rebalance $typeName\n\t" +
       s"current:${ currentShardAllocations.mkString("\n\t\t", "\n\t\t", "") }\n\t" +

diff --git a/src/main/scala/com/evolutiongaming/cluster/MappedAllocationStrategyDDProxy.scala b/src/main/scala/com/evolutiongaming/cluster/MappedAllocationStrategyDDProxy.scala
@@ -27,7 +27,7 @@ class MappedAllocationStrategyDDProxy extends Actor with ActorLogging {
 
   implicit lazy val node = Cluster(context.system)
   lazy val replicator: ActorRef = DistributedData(context.system).replicator
-  private val emptyMap = LWWMap.empty[String, ActorRef]
+  val emptyMap = LWWMap.empty[String, ActorRef]
 
   replicator ! Subscribe(MappingKey, self)