//设置package包名称以及导入依赖的类
package org.http4s.akka

import scala.reflect.ClassTag

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, OneForOneStrategy, Props, Terminated}
import fs2.Task
import fs2.async.mutable.{Queue, Signal}

private class ClientActor[Out](props: Props, outQueue: Queue[Task, Out], closeSignal: Signal[Task, Boolean])
                              (implicit messageType: ClassTag[Out]) extends Actor {
  val serverActor = context actorOf props
  context watch serverActor
  
  def receive: Receive = {
    case Terminated(`serverActor`) =>
      closeSignal.set(true).unsafeRun()
      context stop self
    case messageType(m) if sender() == serverActor =>
      outQueue.enqueue1(m).unsafeRun()
    case m if sender() == serverActor =>
      org.log4s.getLogger.error(s"Server sent unhandled message ${m.getClass.getSimpleName} " +
        s"expecting a ${messageType.runtimeClass.getSimpleName}!")
    case m if sender() == self =>
      serverActor ! m
  }
  
  override def supervisorStrategy: OneForOneStrategy = OneForOneStrategy() {
    case _ => Stop
  }
} 

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

import akka.actor.SupervisorStrategy.{Restart, Stop}
import akka.actor.{Actor, ActorLogging, ActorRef, OneForOneStrategy}


class CustomOneForOneUser extends Actor with ActorLogging {

  // Map that keeps track of how often a given child has been restarted
  var restarts = Map.empty[ActorRef, Int].withDefaultValue(0)

  // Using the default parameters for a SupervisorStrategy mean "arbitarily
  // often over forever"
  override val supervisorStrategy = OneForOneStrategy() {

    case _: ArithmeticException =>

      restarts(sender) match {
        case tooManyRestarts if tooManyRestarts > 15 =>
          restarts -= sender
          Stop
        case n =>
          restarts = restarts.updated(sender, n+1)
          Restart
      }
  }

  override def receive: Receive = ???
} 

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

import akka.actor.SupervisorStrategy.{Escalate, Restart, Stop}
import akka.actor.{Actor, ActorLogging, AllForOneStrategy}

import scala.concurrent.duration._


class ProcrusteanManager extends Actor with ActorLogging {

  override val supervisorStrategy =
                                  AllForOneStrategy(maxNrOfRetries = 5,
                                                    withinTimeRange = 5 seconds,
                                                    loggingEnabled = true) {
    case _: IllegalStateException => Restart
    case _: IllegalArgumentException => Stop
    case _: NullPointerException => Stop
    case _: Exception => Escalate
  }

  override def receive: Receive = ???
} 

//设置package包名称以及导入依赖的类
package net.ruippeixotog.scalafbp.runtime

import akka.actor.SupervisorStrategy.{ Decider, Stop }
import akka.actor._
import NetworkBrokerSupervisorStrategy._

class NetworkBrokerSupervisorStrategy(onError: (ActorRef, Throwable) => Unit)
    extends OneForOneStrategy()(stoppingDecider) {

  override def handleFailure(
    context: ActorContext,
    child: ActorRef,
    cause: Throwable,
    stats: ChildRestartStats,
    children: Iterable[ChildRestartStats]): Boolean = {
    cause match {
      case _: ActorKilledException | _: DeathPactException =>
      case ex: Exception => onError(child, cause)
    }
    super.handleFailure(context, child, cause, stats, children)
  }

  override val loggingEnabled = false
}

object NetworkBrokerSupervisorStrategy {
  def stoppingDecider: Decider = {
    case _: Exception => Stop
  }
} 

//设置package包名称以及导入依赖的类
package io.ticofab.scalarabbitmqexample.actor

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, OneForOneStrategy, Props}
import io.ticofab.scalarabbitmqexample.actor.QueueListener.{CloseYourEars, Listen}
import io.ticofab.scalarabbitmqexample.actor.Supervisor.{Begin, End}

object Supervisor {

  case object Begin

  case object End

  def props = Props[Supervisor]
}

class Supervisor extends Actor {
  val queueListener = context.actorOf(QueueListener.props)

  // very simple supervision strategy: if anything happens, stop the actor
  override val supervisorStrategy = OneForOneStrategy(loggingEnabled = false) {
    case _: Exception => Stop
  }

  override def receive: Receive = {
    case Begin => queueListener ! Listen
    case End => queueListener ! CloseYourEars
  }
} 

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

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, ActorLogging}
import com.actorsys.service._


class RemoteActor extends Actor with ActorLogging {
  val SUCCESS = "SUCCESS"
  val FAILURE = "FAILURE"

  override def receive: Receive = {
    case Start => log.info("receive event" + Start)
    case Stop => log.info("receive event" + Stop)
    case Shutdown(waitSecs) => {
      log.info("wait to shutdown:waitSecs:" + waitSecs)
      Thread.sleep(waitSecs)
      log.info("Shutdown this system.")
      context.system.deadLetters
    }
    case Heartbeat(id, magic) => log.info("receive Heartbeat:" + (id, magic))
    case Header(id, len, encrypted) => log.info("receive Header:" + (id, len, encrypted))
    case Packet(id, seq, content) => {
      val originalSender = sender
      log.info("receive Packet:" + (id, seq, content))
      originalSender ! (seq, SUCCESS)
    }
    case _ => log.info("ERROR")
  }
} 

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

import akka.actor.SupervisorStrategy.{Restart, Stop}
import akka.actor._
import com.bob.akka.supervision.FlakyExpressionCalculator.{FlakinessException, Result}

class ArithmeticService extends Actor with ActorLogging {

  override def receive: Receive = {
    case e: Expression =>
      val worker = context.actorOf(FlakyExpressionCalculator.props(expr = e, position = FlakyExpressionCalculator.Left))
      pendingWorkers += worker -> sender()
    case Result(originalExpression, value, _) =>
      notifyConsumerSuccess(worker = sender(), result = value)
  }

  var pendingWorkers = Map[ActorRef, ActorRef]()

  def notifyConsumerFailure(worker: ActorRef, failure: Throwable): Unit = {
    pendingWorkers.get(worker).foreach { x => x ! Status.Failure(failure) }
    pendingWorkers -= worker
  }

  def notifyConsumerSuccess(worker: ActorRef, result: Int): Unit = {
    pendingWorkers.get(worker) foreach {
      _ ! result
    }
    pendingWorkers -= worker
  }

  override val supervisorStrategy = OneForOneStrategy(loggingEnabled = false) {
    case _: FlakinessException =>
      log.warning("Evaluation of a top level expression failed, restarting.")
      Restart
    case e: ArithmeticException =>
      log.error("Evaluation failed because of: {}", e.getMessage)
      notifyConsumerFailure(worker = sender(), failure = e)
      Stop
    case e =>
      log.error("Unexpected failure: {}", e.getMessage)
      notifyConsumerFailure(worker = sender(), failure = e)
      Stop
  }

} 

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

import actors.WorkflowLog.LogMessage
import akka.actor.SupervisorStrategy.{Restart, Stop}
import akka.actor.{Actor, ActorLogging, OneForOneStrategy}
import com.amazonaws.{AmazonClientException, AmazonServiceException}

import scala.concurrent.duration._

trait AWSSupervisorStrategy extends Actor with ActorLogging {
  override val supervisorStrategy = OneForOneStrategy(maxNrOfRetries = 20, loggingEnabled = true) {
    case ex: AmazonServiceException =>
      ex.getErrorCode match {
        case "ServiceUnavailable" | "Throttling" =>
          log.debug("Supervisor Authorized Restart")
          Restart
        case _ =>
          context.parent ! LogMessage(ex.toString)
          Stop
      }
      
    case _: AmazonClientException =>
      log.debug("Supervisor Authorized Restart")
      Restart

    case ex: Exception =>
      context.parent ! LogMessage(ex.toString)
      Stop
  }
} 

//设置package包名称以及导入依赖的类
package dcp.business.actors

import BoDaoCommon.Log.LogTool
import akka.actor.Actor
import akka.actor.SupervisorStrategy.Stop
import dcp.common._

import scala.collection.mutable.ArrayBuffer
import scala.concurrent.duration._
import dcp.config.DynConfiguration

case class FlushLog()

class LogActor extends Actor {
  private lazy val logs = ArrayBuffer[LogMsg]()
  
  private def cachelogs(logMsg: LogMsg) = {
    logs += logMsg
  }

  private def stopLoger() = {
    this.flushlogs()  //?????
    context.stop(self)
  }

  def receive = {
    case log @ flushLog() => flushlog(log)  //??????????????

    case log @ cacheLog() => cachelogs(log)

    case DeleteLog(lever, date) =>
      lever match {
        case LogLevelEnum.Info => LogTool.deleteInfoLog(date)
        case LogLevelEnum.Warn => LogTool.deleteWarningLog(date)
        case LogLevelEnum.Err => LogTool.deleteErrorLog(date)
      }

    case DeleteDayLogs(date) => LogTool.deleteDayLog(date)

    case FlushLog() =>
      this.flushlogs()
      if (logs.nonEmpty) logs.clear()  //????

    case Stop => this.stopLoger()
  }

} 

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

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, ActorLogging, ActorRef, OneForOneStrategy, Props, Terminated}
import com.example.coalmine.Miner.{DoValuableWork, WorkDone}



object CrookedForeman {
  val props = Props[CrookedForeman]

  case object HireMiners
  case object BerateMiners
  case class DeathNotice(miner: ActorRef)
}

class CrookedForeman extends Actor with ActorLogging {
  import CrookedForeman._

  // We'll learn more about supervisor strategies next week... for now we
  // install this one so that dead children stay dead, to make our point.
  override val supervisorStrategy = OneForOneStrategy() {
    case _: Exception => Stop
  }

  override def receive: Receive = {

    case HireMiners => 1 to 5 foreach { i => {
                                        val miner = context.actorOf(Miner.props,
                                          s"Miner-no-$i")
                                        log.info(s"Hiring miner $i")
                                        context.watch(miner) }
                                      }

    case BerateMiners =>
      val numMiners = context.children.size
      log.info(s"I have $numMiners miner children, at your service! Berating!")
      context.children.foreach { miner => miner ! DoValuableWork }

    case WorkDone(v) => context.parent forward WorkDone(v)

    case Terminated(miner) =>
      log.info(s"Oh dear, $miner has died!")
      context.parent ! DeathNotice(miner)
  }
} 

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

import akka.actor.Actor
import akka.actor.ActorRef
import akka.actor.SupervisorStrategy
import akka.actor.OneForOneStrategy
import akka.event.LoggingReceive
import akka.actor.Props
import auctionsystem.Notifier.Notification
import auctionsystem.Notifier.RepeatRequest
import auctionsystem.exceptions.RemoteServerErrorException
import akka.actor.SupervisorStrategy.Resume
import akka.actor.SupervisorStrategy.Stop

class Notifier extends Actor {
  val auctionPublisher = context.actorSelection("akka.tcp://[email protected]:2552/user/auctionPublisher")
  var requestsMap = Map [ActorRef, Notification]()
  var idx = 0;
  override def supervisorStrategy: SupervisorStrategy = OneForOneStrategy(maxNrOfRetries = 3) {
     case _: RemoteServerErrorException =>
       println(s"An error occurred when trying to connect with remote server, node that caused failure: ${sender.path}. Resuming...")
       self ! RepeatRequest(requestsMap(sender), sender)
       Resume
 
     case e =>
       println(s"Something else went wrong: $e")
       Stop
  }
  
  def receive = LoggingReceive {
    case Notification(auction, highestBidder, highestBid) =>
      val notifierRequest = context.actorOf(Props(new NotifierRequest(auctionPublisher)), "notifierRequest" + idx);      
      idx = idx + 1
      val notification = Notification(auction, highestBidder, highestBid)
      requestsMap += (notifierRequest -> notification)
      notifierRequest ! notification
  }
}

object Notifier {
  final case class RepeatRequest(notification: Notification, notifierRequest: ActorRef)
  final case class Notification(auctionName: String, highestBidder: ActorRef, highestBid: Int);
} 

//设置package包名称以及导入依赖的类
import akka.actor.SupervisorStrategy.Stop
import akka.actor.{Actor, OneForOneStrategy, SupervisorStrategy}

import scala.concurrent.duration.Duration
import scala.io.Source._

class SearchDatabaseActor(filename: String) extends Actor {
  override def receive: Receive = {
    case Search(title) =>
      val lines = fromFile(filename).getLines
      sender ! SearchResponse(lines.filter(s => s.startsWith(title)).toArray)
  }

  override def supervisorStrategy: SupervisorStrategy =
    OneForOneStrategy(10, Duration(60, "seconds")) {
      case _: Exception => Stop
    }
}

case class SearchResponse(books: Array[String]) 

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

import akka.actor.SupervisorStrategy.{Resume, Restart, Stop, Escalate}
import akka.actor.{OneForOneStrategy, Actor, ActorLogging, Props}
import com.typesafe.config.Config
import play.api.libs.ws.WSClient
import scala.concurrent.duration._

class BotSupervisor(wsClient: WSClient,config: Config) extends Actor with ActorLogging{

  override val supervisorStrategy =
    OneForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 minute) {
      case _: ArithmeticException      => Resume
      case _: NullPointerException     => Restart
      case _: IllegalArgumentException => Stop
      case _: Exception                => Escalate
    }

  val tinderBot =
    context
      .actorOf(Props(new TinderBot(wsClient,config))
        ,"TinderBot")

  override def preStart() = {
    log.info("Starting Tinder Bot Supervisor")
        tinderBot ! TinderBot.Start
  }



  override def receive: Receive = {
    case _ =>
  }
}

object BotSupervisor {
  def props = Props[BotSupervisor]
} 

//设置package包名称以及导入依赖的类
package trove.core.event

import akka.actor.SupervisorStrategy.{Resume, Stop}
import akka.actor.{ActorInitializationException, ActorKilledException, OneForOneStrategy, SupervisorStrategy, SupervisorStrategyConfigurator}
import grizzled.slf4j.Logging

class AkkaSupervision extends SupervisorStrategyConfigurator with Logging {
  override def create(): SupervisorStrategy = {
    logger.debug("Creating custom akka supervisor strategy")
    OneForOneStrategy() {
      case _: ActorInitializationException => Stop
      case _: ActorKilledException         => Stop
      case e: Exception                    =>
        logger.error("Exception in event listener", e)
        Resume
    }    
  }
} 

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

import akka.actor.{Actor, AllForOneStrategy, Props}
import akka.actor.SupervisorStrategy.{Escalate, Restart, Resume, Stop}
import com.nossin.ndb.messages.Calculator.{Div, Sub}
//import com.nossin.ndb.messages.Calculator.{Add, Div, Sub}
import scala.concurrent.duration._


class SupervisorAllActor extends Actor{
    override val supervisorStrategy =
      AllForOneStrategy(maxNrOfRetries = 10, withinTimeRange = 1 seconds)
      {
        case _: ArithmeticException => Restart
        case _: NullPointerException => Resume
        case _: IllegalArgumentException => Stop
        case _: Exception => Escalate

      }
    val printer = context.actorOf(Props[ResultPrinterActor])
    val calculator =  context.actorOf(Props(classOf[CalculatorActor], printer)) //The all supervisor strategy will also restart printer!
    def receive = {
      case "Start" => calculator ! com.nossin.ndb.messages.Calculator.Add(10, 12)
        calculator ! Sub(12, 10)
        calculator ! Div(5, 2)
        calculator ! Div(5, 0)
    }
} 

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

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{OneForOneStrategy, SupervisorStrategy, Props, Actor}

sealed trait FuseStatus
case object Intact extends FuseStatus
case object Blown extends FuseStatus

class Fuse(childSpec: Props) extends Actor {
  case object FuseBlown
  var status: FuseStatus = Intact
  val child = context.actorOf(childSpec)

  def receive = {
    case Fuse.ReadStatus => sender() ! status
    case FuseBlown => status = Blown
    case x if status == Intact => child forward x
  }

  override def supervisorStrategy = OneForOneStrategy() {
    case _: Throwable =>
      self ! FuseBlown
      Stop
  }
}

object Fuse {
  case object ReadStatus
  def props(childSpec: Props) = Props(new Fuse(childSpec))
} 

//设置package包名称以及导入依赖的类
import CounterService._
import Worker._
import akka.actor.SupervisorStrategy.Stop
import akka.actor._
import akka.event.LoggingReceive
import akka.pattern.{ask, pipe}
import akka.util.Timeout

import scala.concurrent.duration.{Duration, _}


class Worker extends Actor with ActorLogging {

  implicit val askTimeout = Timeout(5 seconds)

  // Stop the CounterService child if it throws ServiceUnavailable
  override val supervisorStrategy = OneForOneStrategy() {
    case _: CounterService.ServiceUnavailable => Stop
  }

  // The sender of the initial Start message will continuously be notified
  // about progress
  var progressListener: Option[ActorRef] = None
  val counterService = context.actorOf(Props[CounterService], name = "counter")
  val totalCount = 51

  import context.dispatcher

  // Use this Actors' Dispatcher as ExecutionContext

  def receive = LoggingReceive {
    case Start if progressListener.isEmpty =>
      progressListener = Some(sender())
      context.system.scheduler.schedule(Duration.Zero, 1 second, self, Do)

    case Do =>
      counterService ! Increment(1)
      counterService ! Increment(1)
      counterService ! Increment(1)

      // Send current progress to the initial sender
      counterService ? GetCurrentCount map {
        case CurrentCount(_, count) => Progress(100.0 * count / totalCount)
      } pipeTo progressListener.get
  }
}

object Worker {

  case object Start

  case object Do

  final case class Progress(percent: Double)

} 

//设置package包名称以及导入依赖的类
package io.scalanator.robot.monitor

import akka.actor.SupervisorStrategy.Stop
import akka.actor.{ Actor, ActorLogging, ActorSystem, OneForOneStrategy, Props }

final case class ShutdownRequest(reason: String, isError: Boolean = false)

class MonitorActor(mainProps: Props, shutdownFn: (ActorSystem, ShutdownRequest) => Unit) extends Actor with ActorLogging {

  override val supervisorStrategy = OneForOneStrategy() {
    case ex: Exception =>
      log.error(s"Error seen  by monitor actor ${ex.getMessage}", ex)
      self ! ShutdownRequest(s"Monitor actor failed with ${ex.getClass} - ${ex.toString}", isError = true)
      Stop
  }

  def initialiseChildren(): Unit = {

    context.actorOf(mainProps, "robot")

    //    Environment.info foreach log.info
  }

  override def preStart(): Unit = {
    try {
      initialiseChildren()
    } catch {
      case t: Throwable =>
        log.error(s"Error during startup - ${t.getMessage}", t)
        self ! ShutdownRequest(t.toString, isError = true)
    }
  }
  override def receive: Receive = {
    case req: ShutdownRequest =>
      triggerShutdown(req)
  }

  def triggerShutdown(request: ShutdownRequest): Unit = {
    shutdownFn(context.system, request)
  }
} 

//设置package包名称以及导入依赖的类
package teleporter.integration.akka

import akka.actor.SupervisorStrategy.{Decider, Restart, Stop}
import akka.actor._
import org.apache.logging.log4j.scala.Logging


trait FaultHandling {

}

object SupervisorStrategy extends Logging {
  final val defaultDecider: Decider = {
    case e ?
      logger.info(e.getLocalizedMessage, e)
      e match {
        case _: ActorInitializationException ? Stop
        case _: ActorKilledException ? Stop
        case _: DeathPactException ? Stop
        case _: Exception ? Restart
      }
  }

  final val defaultStrategy: SupervisorStrategy = {
    OneForOneStrategy()(defaultDecider)
  }
}

final class DefaultSupervisorStrategy extends SupervisorStrategyConfigurator {
  override def create(): SupervisorStrategy = SupervisorStrategy.defaultStrategy
} 

//设置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
      })
  }
}