<commit_before>/*
 * Copyright (C) 2016-2019 Lightbend Inc. <https://www.lightbend.com>
 */

package com.lightbend.lagom.internal.persistence.cluster

import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Address
import akka.actor.BootstrapSetup
import akka.actor.Props
import akka.actor.setup.ActorSystemSetup
import akka.cluster.Cluster
import akka.cluster.MemberStatus
import akka.cluster.sharding.ClusterShardingSettings
import akka.remote.testconductor.RoleName
import akka.remote.testkit.MultiNodeConfig
import akka.remote.testkit.MultiNodeSpec
import akka.testkit.ImplicitSender
import akka.testkit.TestProbe
import com.lightbend.lagom.internal.persistence.cluster.ClusterDistribution.EnsureActive
import com.lightbend.lagom.scaladsl.persistence.multinode.STMultiNodeSpec
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._

object ClusterDistributionConfig extends MultiNodeConfig {
  val node1 = role("node1")
  val node2 = role("node2")
  val node3 = role("node3")
}

// heavily inspired by AbstractClusteredPersistentEntitySpec
object ClusterDistributionSpec {
  // Copied from MultiNodeSpec
  private def getCallerName(clazz: Class[_]): String = {
    val s = Thread.currentThread.getStackTrace.map(_.getClassName).drop(1).dropWhile(_.matches(".*MultiNodeSpec.?$"))
    val reduced = s.lastIndexWhere(_ == clazz.getName) match {
      case -1 => s
      case z  => s.drop(z + 1)
    }
    reduced.head.replaceFirst(""".*\.""", "").replaceAll("[^a-zA-Z_0-9]", "_")
  }
  def createActorSystem(): (Config) => ActorSystem = { config =>
    val testConfig = ConfigFactory.parseString("akka.actor.provider = cluster").withFallback(config)
    val setup      = ActorSystemSetup(BootstrapSetup(ConfigFactory.load(testConfig)))
    ActorSystem(getCallerName(classOf[MultiNodeSpec]), setup)
  }
}

class ClusterDistributionSpecMultiJvmNode1 extends ClusterDistributionSpec
class ClusterDistributionSpecMultiJvmNode2 extends ClusterDistributionSpec
class ClusterDistributionSpecMultiJvmNode3 extends ClusterDistributionSpec

class ClusterDistributionSpec
    extends MultiNodeSpec(ClusterDistributionConfig, ClusterDistributionSpec.createActorSystem())
    with STMultiNodeSpec
    with ImplicitSender {

import ClusterDistributionConfig._
  override def initialParticipants: Int = roles.size


  def join(from: RoleName, to: RoleName): Unit = {
    runOn(from) {
      Cluster(system).join(node(to).address)
    }
    enterBarrier(from.name + "-joined")
  }

  def fullAddress(ref: ActorRef): Address =
    if (ref.path.address.hasLocalScope) Cluster(system).selfAddress
    else ref.path.address

  protected override def atStartup() {
    roles.foreach(n => join(n, node1))
    within(15.seconds) {
      awaitAssert(Cluster(system).state.members.size should be(3))
      awaitAssert(
        Cluster(system).state.members.toIndexedSeq.map(_.status).distinct should be(IndexedSeq(MemberStatus.Up))
      )
    }

    enterBarrier("startup")
  }


  val distributionSettings =
    ClusterDistributionSettings(system)
      .copy(ensureActiveInterval = 1.second)

  "A ClusterDistribution" must {

    "distribute" in {

      val probe = TestProbe()

      val typeName     = "CDTest"
      val props: Props = FakeActor.props(probe.ref)
      val tagNames     = (1 to 10).map(i => s"test-tagName$i").toSet
      ClusterDistribution(system)
        .start(
          typeName,
          props,
          tagNames,
          distributionSettings
        )

      val echoMessage = probe.receiveOne(10.second).asInstanceOf[String]
      echoMessage should include ("test-tagName")

    }
  }
}

import akka.actor.{ Actor, Props }

object FakeActor {
  def props(ref:ActorRef): Props = Props(new FakeActor(ref))
}

class FakeActor(ref:ActorRef) extends Actor {
  override def receive = {
    case EnsureActive(tagName) => ref ! tagName
  }
}<commit_msg>Adopt coding style from AbstractClusteredPersistenceEntity<commit_after>/*
 * Copyright (C) 2016-2019 Lightbend Inc. <https://www.lightbend.com>
 */

package com.lightbend.lagom.internal.persistence.cluster

import akka.actor.ActorRef
import akka.actor.ActorSystem
import akka.actor.Address
import akka.actor.BootstrapSetup
import akka.actor.Props
import akka.actor.setup.ActorSystemSetup
import akka.cluster.Cluster
import akka.cluster.MemberStatus
import akka.remote.testconductor.RoleName
import akka.remote.testkit.MultiNodeConfig
import akka.remote.testkit.MultiNodeSpec
import akka.testkit.ImplicitSender
import akka.testkit.TestProbe
import com.lightbend.lagom.internal.persistence.cluster.ClusterDistribution.EnsureActive
import com.lightbend.lagom.scaladsl.persistence.multinode.STMultiNodeSpec
import com.typesafe.config.Config
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._

object ClusterDistributionConfig extends MultiNodeConfig {
  val node1 = role("node1")
  val node2 = role("node2")
  val node3 = role("node3")

  commonConfig(
      ConfigFactory
        .parseString(
          """
      terminate-system-after-member-removed = 60s

      # increase default timeouts to leave wider margin for Travis.
      # 30s to 60s
      akka.testconductor.barrier-timeout=60s

      akka.test.single-expect-default = 15s

      # Don't terminate the actor system when doing a coordinated shutdown
      akka.coordinated-shutdown.terminate-actor-system = off
      akka.coordinated-shutdown.run-by-jvm-shutdown-hook = off
      akka.cluster.run-coordinated-shutdown-when-down = off

      # multi-jvm tests forms the cluster programmatically
      # therefore we disable Akka Cluster Bootstrap
      lagom.cluster.bootstrap.enabled = off

      # no jvm exit on tests
      lagom.cluster.exit-jvm-when-system-terminated = off

      akka.cluster.sharding.waiting-for-state-timeout = 5s
    """
        )
        .withFallback(ConfigFactory.parseResources("play/reference-overrides.conf"))
    )

}

// heavily inspired by AbstractClusteredPersistentEntitySpec
object ClusterDistributionSpec {
  // Copied from MultiNodeSpec
  private def getCallerName(clazz: Class[_]): String = {
    val s = Thread.currentThread.getStackTrace.map(_.getClassName).drop(1).dropWhile(_.matches(".*MultiNodeSpec.?$"))
    val reduced = s.lastIndexWhere(_ == clazz.getName) match {
      case -1 => s
      case z  => s.drop(z + 1)
    }
    reduced.head.replaceFirst(""".*\.""", "").replaceAll("[^a-zA-Z_0-9]", "_")
  }
  def createActorSystem(): (Config) => ActorSystem = { config =>
    val setup      = ActorSystemSetup(BootstrapSetup(ConfigFactory.load(config)))
    ActorSystem(getCallerName(classOf[MultiNodeSpec]), setup)
  }
}

class ClusterDistributionSpecMultiJvmNode1 extends ClusterDistributionSpec
class ClusterDistributionSpecMultiJvmNode2 extends ClusterDistributionSpec
class ClusterDistributionSpecMultiJvmNode3 extends ClusterDistributionSpec

class ClusterDistributionSpec
    extends MultiNodeSpec(ClusterDistributionConfig, ClusterDistributionSpec.createActorSystem())
    with STMultiNodeSpec
    with ImplicitSender {

  import ClusterDistributionConfig._
  override def initialParticipants: Int = roles.size

  def join(from: RoleName, to: RoleName): Unit = {
    runOn(from) {
      Cluster(system).join(node(to).address)
    }
    enterBarrier(from.name + "-joined")
  }

  def fullAddress(ref: ActorRef): Address =
    if (ref.path.address.hasLocalScope) Cluster(system).selfAddress
    else ref.path.address

  protected override def atStartup() {
    roles.foreach(n => join(n, node1))
    within(15.seconds) {
      awaitAssert(Cluster(system).state.members.size should be(3))
      awaitAssert(
        Cluster(system).state.members.toIndexedSeq.map(_.status).distinct should be(IndexedSeq(MemberStatus.Up))
      )
    }

    enterBarrier("startup")
  }

  //  -------- Before this line all code is testing scaffolding or cluster formation  ------------------

  val distributionSettings =
    ClusterDistributionSettings(system)
      .copy(ensureActiveInterval = 1.second)

  "A ClusterDistribution" must {

    "distribute the entityIds across nodes (so all nodes get a response)" in {

      val probe = TestProbe()

      val typeName     = "CDTest"
      // There'll be 3 nodes in this test and each node will host a TestProbe.
      // Cluster Distribution on each node will create a `FakeActor.props` pointing
      // back to its own TestProbe. We request the creation of 10 FakeActor to ClusterDistribution
      // with the hope that there'll be at least one for each TestProbe. This is flaky but unlikely to fail.
      // If it failed, just increase the number of entityIds.
      val props: Props = FakeActor.props(probe.ref)
      val entityIds     = (1 to 10).map(i => s"test-entityId$i").toSet
      ClusterDistribution(system)
        .start(
          typeName,
          props,
          entityIds,
          distributionSettings
        )

      val echoMessage = probe.receiveOne(10.second).asInstanceOf[String]
      echoMessage should include("test-entityId")

    }
  }
}

import akka.actor.Actor
import akka.actor.Props

object FakeActor {
  def props(ref: ActorRef): Props = Props(new FakeActor(ref))
}

class FakeActor(ref: ActorRef) extends Actor {
  override def receive = {
    case EnsureActive(entityId) => ref ! entityId
  }
}
{'fraction_non_alphanumeric': 0.10143846563665423, 'fraction_numerical': 0.008311134789557806, 'mean_word_length': 3.7912200102092903, 'pattern_counts': {'":': 0, '<': 5, '<?xml version=': 0, '>': 19, 'https://': 2, 'lorem ipsum': 0, 'www.': 2, 'xml': 0}, 'pii_count': 0, 'substrings_counts': 0, 'word_list_counts': {'cursed_substrings.json': 0, 'profanity_word_list.json': 0, 'sexual_word_list.json': 0, 'zh_pornsignals.json': 0}}
{'dir': 'git-commits-cleaned', 'id': '540319', 'n_tokens_mistral': 2905, 'n_tokens_neox': 2733, 'n_words': 606}