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(_))

}