scala akka

Test.scala

package com.lightbend.akka.sample

import akka.actor.{Actor, ActorLogging, ActorSystem, Props}
import akka.routing.RoundRobinPool

object SampleData {
  case class Account(id: Int, name: String)
  case class Campaign(id: Int, accountId: Int, name: String)
  case class Keyword(id: Int, campaignId: Int, name: String)

  val accounts: List[Account] = List(
    Account(1, "Maria"),
    Account(2, "Jon"),
    Account(3, "Snow")
  )

  val campaigns: List[Campaign] = List(
    Campaign(1, 1, "Maria-campaign-1"),
    Campaign(2, 1, "Maria-campaign-2"),
    Campaign(3, 2, "Jon-campaign-1"),
    Campaign(4, 3, "Snow-campaign-1"),
  )

  val keywords: List[Keyword] = List(
    Keyword(1, 1, "A"),
    Keyword(2, 1, "B"),
    Keyword(3, 1, "C"),
    Keyword(4, 1, "D"),
    Keyword(5, 1, "E"),
    Keyword(6, 2, "F"),
    Keyword(7, 2, "G"),
    Keyword(8, 2, "H"),
    Keyword(9, 3, "I"),
  )
}

object Test extends App {
  import com.lightbend.akka.sample.Test.AccountActor._
  import SampleData._

  object AccountActor {
    def props = Props[AccountActor]
    case class ProcessAccount(account: Account)
  }

  class AccountActor extends Actor with ActorLogging {
    import AccountActor._
    import CampaignActor._

    val campaignActor = context.actorOf(CampaignActor.props, "adGroup")

    override def receive: Receive = {
      case ProcessAccount(account) =>
        val campaignIn = for (campaign <- campaigns; if campaign.accountId == account.id) yield campaign
        campaignIn foreach (campaign => campaignActor ! ProcessCampaign(campaign))
    }
  }

  object CampaignActor {
    def props: Props = Props[CampaignActor]
    case class ProcessCampaign(campaign: Campaign)
  }

  class CampaignActor extends Actor with ActorLogging {
    import CampaignActor._
    import KeywordActor._

    val keywordActor = context.actorOf(RoundRobinPool(5).props(KeywordActor.props), "keyword")

    override def receive: Receive = {
      case ProcessCampaign(campaign) =>
        log.info(s"Insert campaign ${campaign.name}")
        val keywordsIn = for (keyword <- keywords; if keyword.campaignId == campaign.id) yield keyword
        keywordsIn foreach (keyword => keywordActor ! ProcessKeyword(keyword))
    }
  }

  object KeywordActor {
    def props = Props[KeywordActor]
    case class ProcessKeyword(keyword: Keyword)
  }

  class KeywordActor extends Actor with ActorLogging {
    import KeywordActor._

    override def receive: Receive = {
      case ProcessKeyword(keyword) =>
        log.info(s"Insert keyword ${keyword.name}")
    }
  }

  val system: ActorSystem = ActorSystem("BatchAkka")
  val accountActor = system.actorOf(AccountActor.props, "account")

  accounts.foreach(accountActor ! ProcessAccount(_))
}