<commit_before>package demo

import akka.actor.typed.{ActorRef, Behavior}
import akka.actor.typed.scaladsl.Behaviors
import akka.cluster.Cluster
import akka.cluster.ddata.{PNCounterMap, PNCounterMapKey}
import akka.cluster.ddata.typed.scaladsl.Replicator.{Command, ModifyFailure, StoreFailure, UpdateDataDeleted, UpdateSuccess, UpdateTimeout}
import akka.cluster.ddata.typed.scaladsl.{Replicator, ReplicatorSettings}
import com.typesafe.config.{Config, ConfigFactory}
import akka.actor.typed.scaladsl.adapter._
import akka.cluster.ddata.typed.scaladsl.Replicator._
import akka.cluster.ddata.typed.scaladsl.ReplicatorMessageAdapter
import demo.RingMaster.PingDeviceReply

import scala.concurrent.duration._
import scala.concurrent.duration.FiniteDuration

object ShardReplicator {

  sealed trait Protocol

  final case class PingReplicator(deviceId: Long, replyTo: ActorRef[PingDeviceReply]) extends Protocol

  private final case class InternalUpdateResponse(
    rsp: UpdateResponse[PNCounterMap[Long]],
    replyTo: ActorRef[PingDeviceReply]
  ) extends Protocol

  private case class InternalDataUpdated(chg: Replicator.SubscribeResponse[PNCounterMap[Long]]) extends Protocol

  private val Key = PNCounterMapKey[Long]("device-counters")

  def replicatorConfig(role: String): Config =
    ConfigFactory.parseString(
      s"""
         | role = $role
         |
         | gossip-interval = 1 s
         |
         | use-dispatcher = "akka.actor.default-dispatcher"
         |
         | notify-subscribers-interval = 2000 ms
         |
         | max-delta-elements = 100
         |
         | pruning-interval = 120 s
         |
         | max-pruning-dissemination = 300 s
         |
         | pruning-marker-time-to-live = 6 h
         |
         | serializer-cache-time-to-live = 10s
         |
         | prefer-oldest = off
         |
         | delta-crdt {
         |   enabled = on
         |   max-delta-size = 50
         | }
         |
         | durable {
         |  keys = []
         |  pruning-marker-time-to-live = 10 d
         |  store-actor-class = akka.cluster.ddata.LmdbDurableStore
         |  pinned-store {
         |    type = PinnedDispatcher
         |    executor = thread-pool-executor
         |  }
         |
         |  use-dispatcher = akka.cluster.distributed-data.durable.pinned-store
         | }
         |
        """.stripMargin
    )

  def apply(shardName: String, to: FiniteDuration = 1.seconds): Behavior[Protocol] =
    Behaviors.setup[Protocol] { ctx ⇒
      implicit val cl = Cluster(ctx.system.toClassic)

      //WriteMajority(to) respects in reachable nodes
      val wc = WriteLocal //WriteMajority(to) WriteTo(2, to)

      val ddConf = replicatorConfig(shardName)
      val akkaReplicator: ActorRef[Command] =
        ctx.spawn(akka.cluster.ddata.typed.scaladsl.Replicator.behavior(ReplicatorSettings(ddConf)), "ddata-replicator")

      val adapter = ReplicatorMessageAdapter[Protocol, PNCounterMap[Long]](ctx, akkaReplicator, to)
      adapter.subscribe(Key, InternalDataUpdated.apply)

      def behavior: PartialFunction[Protocol, Behavior[Protocol]] = {
        case PingReplicator(deviceId, replyTo) ⇒
          ctx.log.warn(s"Write key:${Key.id} - device:$deviceId")
          adapter.askUpdate(
            replyTo ⇒ Replicator.Update(Key, PNCounterMap.empty[Long], wc, replyTo)(_.increment(deviceId, 1)),<commit_msg>clean-up<commit_after>package demo

import akka.actor.typed.{ActorRef, Behavior}
import akka.actor.typed.scaladsl.Behaviors
import akka.cluster.Cluster
import akka.cluster.ddata.{PNCounterMap, PNCounterMapKey}
import akka.cluster.ddata.typed.scaladsl.Replicator.{Command, ModifyFailure, StoreFailure, UpdateDataDeleted, UpdateSuccess, UpdateTimeout}
import akka.cluster.ddata.typed.scaladsl.{Replicator, ReplicatorSettings}
import com.typesafe.config.{Config, ConfigFactory}
import akka.actor.typed.scaladsl.adapter._
import akka.cluster.ddata.typed.scaladsl.Replicator._
import akka.cluster.ddata.typed.scaladsl.ReplicatorMessageAdapter
import demo.RingMaster.PingDeviceReply

import scala.concurrent.duration._
import scala.concurrent.duration.FiniteDuration

object ShardReplicator {

  sealed trait Protocol

  final case class PingReplicator(deviceId: Long, replyTo: ActorRef[PingDeviceReply]) extends Protocol

  private final case class InternalUpdateResponse(
    rsp: UpdateResponse[PNCounterMap[Long]],
    replyTo: ActorRef[PingDeviceReply]
  ) extends Protocol

  private case class InternalDataUpdated(chg: Replicator.SubscribeResponse[PNCounterMap[Long]]) extends Protocol

  private val Key = PNCounterMapKey[Long]("device-counters")

  def replicatorConfig(role: String): Config =
    ConfigFactory.parseString(
      s"""
         | role = $role
         |
         | gossip-interval = 1 s
         |
         | use-dispatcher = "akka.actor.default-dispatcher"
         |
         | notify-subscribers-interval = 2000 ms
         |
         | max-delta-elements = 100
         |
         | pruning-interval = 120 s
         |
         | max-pruning-dissemination = 300 s
         |
         | pruning-marker-time-to-live = 6 h
         |
         | serializer-cache-time-to-live = 10s
         |
         | prefer-oldest = off
         |
         | delta-crdt {
         |   enabled = on
         |   max-delta-size = 50
         | }
         |
         | durable {
         |  keys = []
         |  pruning-marker-time-to-live = 10 d
         |  store-actor-class = akka.cluster.ddata.LmdbDurableStore
         |  pinned-store {
         |    type = PinnedDispatcher
         |    executor = thread-pool-executor
         |  }
         |
         |  use-dispatcher = akka.cluster.distributed-data.durable.pinned-store
         | }
         |
        """.stripMargin
    )

  def apply(shardName: String, to: FiniteDuration = 1.seconds): Behavior[Protocol] =
    Behaviors.setup[Protocol] { ctx ⇒
      //WriteMajority(to) WriteTo(2, to)
      implicit val clusterCluster = Cluster(ctx.system.toClassic)

      val ddConf = replicatorConfig(shardName)
      val akkaReplicator: ActorRef[Command] =
        ctx.spawn(akka.cluster.ddata.typed.scaladsl.Replicator.behavior(ReplicatorSettings(ddConf)), "ddata-replicator")

      val adapter = ReplicatorMessageAdapter[Protocol, PNCounterMap[Long]](ctx, akkaReplicator, to)
      adapter.subscribe(Key, InternalDataUpdated.apply)

      def behavior: PartialFunction[Protocol, Behavior[Protocol]] = {
        case PingReplicator(deviceId, replyTo) ⇒
          ctx.log.warn(s"Write key:${Key.id} - device:$deviceId")
          adapter.askUpdate(
            replyTo ⇒ Replicator.Update(Key, PNCounterMap.empty[Long], WriteLocal, replyTo)(_.increment(deviceId, 1)),
{'fraction_non_alphanumeric': 0.1030989272943981, 'fraction_numerical': 0.007151370679380214, 'mean_word_length': 2.7926553672316383, 'pattern_counts': {'":': 0, '<': 3, '<?xml version=': 0, '>': 3, 'https://': 0, 'lorem ipsum': 0, 'www.': 0, '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': '7647661', 'n_tokens_mistral': 2169, 'n_tokens_neox': 2079, 'n_words': 359}