//设置package包名称以及导入依赖的类
package com.example

import akka.actor.{ActorSystem,Props}
import akka.pattern.{Backoff,BackoffSupervisor}
import scala.concurrent.duration._
import com.typesafe.config.ConfigFactory
import kamon.Kamon

object Main {
  def main(args: Array[String]): Unit = {
	  Kamon.start()
    if (args.isEmpty || args.head == "Server")
      startServerSystem()
    if (args.isEmpty || args.head == "Writer")
      startWriterSystem()
    if (args.isEmpty || args.head == "Reader")
      startReaderSystem()
  }
  
  def startServerSystem() = {
    val system = ActorSystem("ServerSystem", ConfigFactory.load("server"))
    import system.dispatcher
    val serverActor = supervise(system, ServerActor.props, "serverActor")
    system.scheduler.schedule(1 second, 1 seconds){serverActor ! "print"}
  }
  
  def startWriterSystem() = {
    val config = ConfigFactory.load("writer")
    val serverHost = config.getString("serverHost")
    val serverPort = config.getInt("serverPort")
    val system = ActorSystem("WriterSystem", config)
    val serverSelection = system.actorSelection(s"akka.tcp://[email protected]$serverHost:$serverPort/user/serverActor")
    val writerActor = supervise(system, WriterActor.props(serverSelection), "writerActor")
  }
  
  def startReaderSystem() = {
    val config = ConfigFactory.load("reader")
    val serverHost = config.getString("serverHost")
    val serverPort = config.getInt("serverPort")
    val system = ActorSystem("ReaderSystem", config)
    import system.dispatcher
    val serverSelection = system.actorSelection(s"akka.tcp://[email protected]$serverHost:$serverPort/user/serverActor")
    val readerActor = supervise(system, ReaderActor.props(serverSelection, 1000), "readerActor")
    system.scheduler.schedule(1 second, 1 seconds){readerActor ! "print"}
  }
  
  def supervise(system: ActorSystem, childProps: Props, name: String) = {
    val supervisor = BackoffSupervisor.props(
      Backoff.onStop(
        childProps,
        childName = name,
        minBackoff = 3 seconds,
        maxBackoff = 30 seconds,
        randomFactor = 0.2 // adds 20% "noise" to vary the intervals slightly
      ))
    system.actorOf(supervisor, name = name)

  }
} 

//设置package包名称以及导入依赖的类
package sample.kamon

import akka.actor._
import akka.pattern.{Backoff, BackoffSupervisor}
import kamon.Kamon
import scala.concurrent.duration._

object Main extends App {

  Kamon.start()

  val system = ActorSystem("application")

  val actorProps = Props[LongConsumer]

  val supervisor = BackoffSupervisor.props(
    Backoff.onStop(
      actorProps,
      childName = "telemetry-consumer",
      minBackoff = 3.seconds,
      maxBackoff = 30.seconds,
      randomFactor = 0.2 // adds 20% "noise" to vary the intervals slightly
    ).withSupervisorStrategy(
      OneForOneStrategy() {
        case ex =>
          system.log.error(ex, "There was an error in KafkaActor")
          SupervisorStrategy.Restart //Here we can add some log or send a notification
      })
  )

  system.actorOf(supervisor)
} 

//设置package包名称以及导入依赖的类
package sample.stream_actor

import akka.actor.ActorSystem
import akka.pattern.{Backoff, BackoffSupervisor}
import akka.stream.{ActorMaterializer, ThrottleMode}
import akka.stream.scaladsl.{Sink, Source}
import sample.WindTurbineSimulator

import scala.concurrent.duration._


object SimulateWindTurbines extends App {
  implicit val system = ActorSystem()
  implicit val materializer = ActorMaterializer()

  val endpoint = "ws://127.0.0.1:8080"
  val numberOfTurbines = 5
  Source(1 to numberOfTurbines)
    .throttle(
      elements = 100, //number of elements to be taken from bucket
      per = 1.second,
      maximumBurst = 100,  //capacity of bucket
      mode = ThrottleMode.shaping
    )
    .map { _ =>
      val id = java.util.UUID.randomUUID.toString

      val supervisor = BackoffSupervisor.props(
        Backoff.onFailure(
          WindTurbineSimulator.props(id, endpoint),
          childName = id,
          minBackoff = 1.second,
          maxBackoff = 30.seconds,
          randomFactor = 0.2
        ))

      system.actorOf(supervisor, name = s"$id-backoff-supervisor")
    }
    .runWith(Sink.ignore)
} 

//设置package包名称以及导入依赖的类
package com.microworkflow

import akka.actor.{Actor, Props}
import akka.pattern.BackoffSupervisor
import akka.stream.{ActorMaterializer, Materializer}

import scala.concurrent.duration._

class BootstrapActor private(propsBuilder: Materializer ? Props) extends Actor {

  override def receive: Receive = {
    case _ ? 
  }

  implicit val materializer = ActorMaterializer()
  val supervisorProps =
    BackoffSupervisor.props(
      propsBuilder(materializer)
      , "rest-router"
      , minBackoff = 5.seconds
      , maxBackoff = 50.seconds
      , randomFactor = 0.11
    )
  context.actorOf(supervisorProps)
}

object BootstrapActor {

  def props(propsBuilder: Materializer ? Props): Props =
    Props(classOf[BootstrapActor], propsBuilder)
} 

//设置package包名称以及导入依赖的类
package org.dohrm.storyline.users

import java.util.concurrent.TimeUnit

import akka.actor.Props
import akka.pattern.BackoffSupervisor
import org.dohrm.auth0.directives.Auth0
import org.dohrm.storyline.users.actors.UserRepository
import org.dohrm.storyline.users.directives.UserSecurity
import org.dohrm.toolkit.context.{ActorContext, FutureContext, JdbcContext}

import scala.concurrent.duration.FiniteDuration


trait UserContext extends UserSecurity {
  self: ActorContext
    with Auth0
    with FutureContext
    with JdbcContext
  =>

  override val userRepositoryActor = as.actorOf(
    BackoffSupervisor.props(
      Props(new UserRepository),
      "crud-user",
      FiniteDuration(3, TimeUnit.SECONDS),
      FiniteDuration(30, TimeUnit.SECONDS),
      0.42
    )
  )


} 

//设置package包名称以及导入依赖的类
package ex1

import java.util.concurrent.atomic.AtomicInteger

import akka.actor._
import akka.pattern.{Backoff, BackoffSupervisor}

import scala.concurrent.Await
import scala.concurrent.duration._

object Main1 {
  val x = new AtomicInteger(0)

  def main(args: Array[String]) {
    val system = ActorSystem("mofu")
    val ahoProp = BackoffSupervisor.props(
      Backoff.onFailure(
        childProps = BackOffSupervisorActor.props,
        childName = "aho",
        minBackoff = 1.milliseconds,
        maxBackoff = 2.seconds,
        randomFactor = 0
        )
    )
    val ref = system.actorOf(ahoProp)
    (1 to 100).foreach { x =>
      ref ! x
      Thread.sleep(100)
    }
    Await.result(system.whenTerminated, Duration.Inf)
  }
}

class BackOffSupervisorActor extends Actor {
  import Main1.x
  def receive: Receive = {
    case _ if x.get() < 10 =>
      x.incrementAndGet()
      throw new RuntimeException(s"restarted ${x.get()} at ${System.currentTimeMillis()}")
    case _ =>
      println("yeah!")
      context.system.terminate()
  }
}
object BackOffSupervisorActor {
  def props = Props(classOf[BackOffSupervisorActor])
} 

//设置package包名称以及导入依赖的类
package audit

import akka.actor.ActorSystem
import akka.http.scaladsl.Http
import akka.http.scaladsl.server.Directives._
import akka.pattern.BackoffSupervisor
import akka.stream.ActorMaterializer
import audit.collector.CollectorApi
import audit.viewer.ViewerApi
import com.datastax.driver.core.Cluster
import com.datastax.driver.core.policies.{ConstantReconnectionPolicy, DefaultRetryPolicy}
import com.typesafe.config.ConfigFactory

import scala.concurrent.duration._

object AuditApp extends App with CollectorApi with ViewerApi with Api {

  implicit val system = ActorSystem("Audit")
  implicit val executionContext = system.dispatcher
  implicit val materializer = ActorMaterializer()

  val settings = new AuditSettings(ConfigFactory.load())
  import settings._

  val retryPolicy = DefaultRetryPolicy.INSTANCE
  val reconnectionPolicy = new ConstantReconnectionPolicy(CassandraReconnectionDelay)
  val cluster = Cluster.builder().
    addContactPoint(CassandraHost).
    withRetryPolicy(retryPolicy).
    withReconnectionPolicy(reconnectionPolicy)

  val nodeSupervisor = BackoffSupervisor.props(AuditNode.props(cluster, settings), "node", 1.second, 2.minutes, 0.2f)

  val auditNode = system.actorOf(nodeSupervisor)

  val collector = auditNode
  val viewer = auditNode

  val route = collectorRoute ~ viewerRoute

  val binding = Http().bindAndHandle(route, AppHost, AppPort)

  println("Up and running ...")

} 

//设置package包名称以及导入依赖的类
package nl.tradecloud.kafka

import java.util.concurrent.atomic.AtomicInteger

import akka.Done
import akka.actor.{ActorRefFactory, ActorSystem, Props, SupervisorStrategy}
import akka.kafka.ProducerSettings
import akka.pattern.BackoffSupervisor
import akka.stream.Materializer
import nl.tradecloud.kafka.command.Publish
import nl.tradecloud.kafka.config.KafkaConfig
import org.apache.kafka.common.serialization.{ByteArraySerializer, StringSerializer}

import scala.concurrent.duration._
import scala.concurrent.{ExecutionContext, Future, Promise}

class KafkaPublisher(system: ActorSystem)(implicit mat: Materializer, context: ActorRefFactory) {
  import KafkaPublisher._

  implicit val dispatcher: ExecutionContext = system.dispatchers.lookup("dispatchers.kafka-dispatcher")

  val kafkaConfig = KafkaConfig(system.settings.config)

  private lazy val publisherId = KafkaClientIdSequenceNumber.getAndIncrement

  private def publisherSettings = {
    val keySerializer = new StringSerializer
    val valueSerializer = new ByteArraySerializer

    ProducerSettings(system, keySerializer, valueSerializer).withBootstrapServers(kafkaConfig.brokers)
  }

  private val publisherProps: Props = KafkaPublisherActor.props(kafkaConfig, publisherSettings)
  private val backoffPublisherProps: Props = BackoffSupervisor.propsWithSupervisorStrategy(
    publisherProps, s"KafkaPublisherActor$publisherId", 3.seconds,
    30.seconds, 1.0, SupervisorStrategy.stoppingStrategy
  )
  private val publishActor = context.actorOf(backoffPublisherProps, s"KafkaBackoffPublisher$publisherId")

  def publish(topic: String, msg: AnyRef): Future[Done] = {
    val completed: Promise[Done] = Promise()

    publishActor ! Publish(topic, msg, completed)

    completed.future
  }

}

object KafkaPublisher {
  private val KafkaClientIdSequenceNumber = new AtomicInteger(1)
} 

//设置package包名称以及导入依赖的类
package com.akkastarting.chapter4

import scala.concurrent.duration._
import akka.actor.SupervisorStrategy.{Escalate, Stop, Restart, Resume}
import akka.actor.{Actor, ActorLogging, Props, OneForOneStrategy}
import akka.pattern.{BackoffOptions, Backoff, BackoffSupervisor}

class Ch4Ping1Actor extends Actor with ActorLogging {
  log.info("Ping1Actor Created, ...")

  val child2Supervisor = BackoffSupervisor.props(createStrategy(Props[Ch4Ping2Actor], "Ping2Actor"))
  val child3Supervisor = BackoffSupervisor.props(createStrategy(Props[Ch4Ping3Actor], "Ping3Actor"))
  val child2 = context.system.actorOf(child2Supervisor, "Ping2ActorWithSupervisor")
  val child3 = context.system.actorOf(child3Supervisor, "Ping3ActorWithSupervisor")

  def receive = {
    case message: Bad =>
      (child2 ! message)(sender)
      (child3 ! message)(sender)
    case message: Good =>
      (child2 ! message)(sender)
      (child3 ! message)(sender)
  }

  def createStrategy(props: Props, name: String): BackoffOptions = {
    
    Backoff.onStop(props,
      childName = name,
      minBackoff = 3.seconds,
      maxBackoff = 30.seconds,
      randomFactor = 0.2)
      .withManualReset
      .withSupervisorStrategy(OneForOneStrategy() {
        case ex: ArithmeticException =>
          Resume
        case ex: NullPointerException =>
          Restart
        case ex: IllegalArgumentException =>
          Stop
        case _ =>
          Escalate
      })
  }
} 

//设置package包名称以及导入依赖的类
package com.carjump

import akka.actor._
import com.carjump.http.ReqParams
import net.ceedubs.ficus.Ficus._
import scala.concurrent.duration._
import akka.actor.SupervisorStrategy.Directive
import akka.pattern.{ Backoff, BackoffSupervisor }

object CacheGuardian {
  def props = Props[CacheGuardian].withDispatcher(Cache.DispatcherName)
}

class CacheGuardian extends Actor with ActorLogging {
  val conf = context.system.settings.config

  val url = conf.as[String]("ws.url")
  val pref = conf.as[String]("ws.path")
  val pullInterval = conf.as[FiniteDuration]("ws.interval")

  val minBackoffInterval = conf.as[FiniteDuration]("ws.min_backoff")
  val maxBackoffInterval = conf.as[FiniteDuration]("ws.max_backoff")

  val decider: PartialFunction[Throwable, Directive] = {
    case ex: Throwable ?
      log.error(ex, "Cache has failed unexpectedly. We will try to recreate it later")
      akka.actor.SupervisorStrategy.Stop
  }

  val props = BackoffSupervisor.props(
    Backoff.onStop(childProps = Cache.props(url, pref, pullInterval), childName = "cache",
      minBackoff = minBackoffInterval, maxBackoff = maxBackoffInterval, randomFactor = 0.2)
      .withSupervisorStrategy(OneForOneStrategy()(decider.orElse(SupervisorStrategy.defaultStrategy.decider))))

  val fetcher = context.actorOf(props)

  override def receive: Receive = {
    case msg: ReqParams ? (fetcher forward msg)
  }
}