Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Simplify backoff supervision API #26156
- Loading branch information
Showing
15 changed files
with
1,180 additions
and
622 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Large diffs are not rendered by default.
Oops, something went wrong.
502 changes: 136 additions & 366 deletions
502
akka-actor/src/main/scala/akka/pattern/BackoffOptions.scala
Large diffs are not rendered by default.
Oops, something went wrong.
322 changes: 111 additions & 211 deletions
322
akka-actor/src/main/scala/akka/pattern/BackoffSupervisor.scala
Large diffs are not rendered by default.
Oops, something went wrong.
69 changes: 69 additions & 0 deletions
69
akka-actor/src/main/scala/akka/pattern/HandleBackoff.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| /* | ||
| * Copyright (C) 2018-2019 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.pattern | ||
|
|
||
| import akka.actor.{ Actor, ActorRef, Props } | ||
| import akka.annotation.InternalApi | ||
| import akka.pattern.internal.{ BackoffOnRestartSupervisor, BackoffOnStopSupervisor } | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| * | ||
| * Implements basic backoff handling for [[BackoffOnRestartSupervisor]] and [[BackoffOnStopSupervisor]]. | ||
| */ | ||
| @InternalApi private[akka] trait HandleBackoff { | ||
| this: Actor ⇒ | ||
| def childProps: Props | ||
| def childName: String | ||
| def reset: BackoffReset | ||
| protected def handleMessageToChild(m: Any): Unit | ||
|
|
||
| var child: Option[ActorRef] = None | ||
| var restartCount = 0 | ||
| var finalStopMessageReceived = false | ||
|
|
||
| import BackoffSupervisor._ | ||
| import context.dispatcher | ||
|
|
||
| override def preStart(): Unit = startChild() | ||
|
|
||
| def startChild(): Unit = if (child.isEmpty) { | ||
| child = Some(context.watch(context.actorOf(childProps, childName))) | ||
| } | ||
|
|
||
| def handleBackoff: Actor.Receive = { | ||
| case StartChild ⇒ | ||
| startChild() | ||
| reset match { | ||
| case AutoReset(resetBackoff) ⇒ | ||
| context.system.scheduler.scheduleOnce(resetBackoff, self, ResetRestartCount(restartCount)) | ||
| case _ ⇒ // ignore | ||
| } | ||
|
|
||
| case Reset ⇒ | ||
| reset match { | ||
| case ManualReset ⇒ restartCount = 0 | ||
| case msg ⇒ unhandled(msg) | ||
| } | ||
|
|
||
| case ResetRestartCount(current) ⇒ | ||
| if (current == restartCount) { | ||
| restartCount = 0 | ||
| } | ||
|
|
||
| case GetRestartCount ⇒ | ||
| sender() ! RestartCount(restartCount) | ||
|
|
||
| case GetCurrentChild ⇒ | ||
| sender() ! CurrentChild(child) | ||
|
|
||
| case msg if child.contains(sender()) ⇒ | ||
| // use the BackoffSupervisor as sender | ||
| context.parent ! msg | ||
|
|
||
| case msg ⇒ | ||
| handleMessageToChild(msg) | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
93 changes: 93 additions & 0 deletions
93
akka-actor/src/main/scala/akka/pattern/internal/BackoffOnStopSupervisor.scala
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,93 @@ | ||
| /* | ||
| * Copyright (C) 2018-2019 Lightbend Inc. <https://www.lightbend.com> | ||
| */ | ||
|
|
||
| package akka.pattern.internal | ||
|
|
||
| import akka.actor.SupervisorStrategy.{ Directive, Escalate } | ||
| import akka.actor.{ Actor, ActorLogging, OneForOneStrategy, Props, SupervisorStrategy, Terminated } | ||
| import akka.annotation.InternalApi | ||
| import akka.pattern.{ BackoffReset, BackoffSupervisor, HandleBackoff } | ||
|
|
||
| import scala.concurrent.duration.FiniteDuration | ||
|
|
||
| /** | ||
| * INTERNAL API | ||
| * | ||
| * Back-off supervisor that stops and starts a child actor using a back-off algorithm when the child actor stops. | ||
| * This back-off supervisor is created by using `akka.pattern.BackoffSupervisor.props` | ||
| * with `BackoffOpts.onStop`. | ||
| */ | ||
| @InternalApi private[pattern] class BackoffOnStopSupervisor( | ||
| val childProps: Props, | ||
| val childName: String, | ||
| minBackoff: FiniteDuration, | ||
| maxBackoff: FiniteDuration, | ||
| val reset: BackoffReset, | ||
| randomFactor: Double, | ||
| strategy: SupervisorStrategy, | ||
| replyWhileStopped: Option[Any], | ||
| finalStopMessage: Option[Any ⇒ Boolean]) | ||
| extends Actor with HandleBackoff | ||
| with ActorLogging { | ||
|
|
||
| import BackoffSupervisor._ | ||
| import context.dispatcher | ||
|
|
||
| override val supervisorStrategy = strategy match { | ||
| case oneForOne: OneForOneStrategy ⇒ | ||
| OneForOneStrategy(oneForOne.maxNrOfRetries, oneForOne.withinTimeRange, oneForOne.loggingEnabled) { | ||
| case ex ⇒ | ||
| val defaultDirective: Directive = | ||
| super.supervisorStrategy.decider.applyOrElse(ex, (_: Any) ⇒ Escalate) | ||
|
|
||
| strategy.decider.applyOrElse(ex, (_: Any) ⇒ defaultDirective) | ||
| } | ||
| case s ⇒ s | ||
| } | ||
|
|
||
| def onTerminated: Receive = { | ||
| case Terminated(ref) if child.contains(ref) ⇒ | ||
| child = None | ||
| if (finalStopMessageReceived) { | ||
| context.stop(self) | ||
| } else { | ||
| val maxNrOfRetries = strategy match { | ||
| case oneForOne: OneForOneStrategy ⇒ oneForOne.maxNrOfRetries | ||
| case _ ⇒ -1 | ||
| } | ||
|
|
||
| val nextRestartCount = restartCount + 1 | ||
|
|
||
| if (maxNrOfRetries == -1 || nextRestartCount <= maxNrOfRetries) { | ||
| val restartDelay = calculateDelay(restartCount, minBackoff, maxBackoff, randomFactor) | ||
| context.system.scheduler.scheduleOnce(restartDelay, self, StartChild) | ||
| restartCount = nextRestartCount | ||
| } else { | ||
| log.debug(s"Terminating on restart #{} which exceeds max allowed restarts ({})", nextRestartCount, maxNrOfRetries) | ||
| context.stop(self) | ||
| } | ||
| } | ||
|
|
||
| } | ||
|
|
||
| def receive: Receive = onTerminated orElse handleBackoff | ||
|
|
||
| protected def handleMessageToChild(msg: Any): Unit = child match { | ||
| case Some(c) ⇒ | ||
| c.forward(msg) | ||
| if (!finalStopMessageReceived) finalStopMessage match { | ||
| case Some(fsm) ⇒ finalStopMessageReceived = fsm(msg) | ||
| case None ⇒ | ||
| } | ||
| case None ⇒ | ||
| replyWhileStopped match { | ||
| case Some(r) ⇒ sender() ! r | ||
| case None ⇒ context.system.deadLetters.forward(msg) | ||
| } | ||
| finalStopMessage match { | ||
| case Some(fsm) if fsm(msg) ⇒ context.stop(self) | ||
| case _ ⇒ | ||
| } | ||
| } | ||
| } |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Oops, something went wrong.