Actor lifecycle

TODO intro

Creating Actors

TODO

Stopping Actors

An actor can stop itself by returning Behaviors.stopped as the next behavior.

Child actors can be forced to be stopped after it finishes processing its current message by using the stop method of the ActorContext from the parent actor. Only child actors can be stopped in that way.

The child actors will be stopped as part of the shutdown procedure of the parent.

The PostStop signal that results from stopping an actor can be used for cleaning up resources. Note that a behavior that handles such PostStop signal can optionally be defined as a parameter to Behaviors.stopped if different actions is needed when the actor gracefully stops itself from when it is stopped abruptly.

Here is an illustrating example:

Scala

import akka.actor.typed.scaladsl.Behaviors
import akka.actor.typed.{ ActorSystem, Logger, PostStop }
import akka.testkit.typed.scaladsl.ActorTestKit

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

object MasterControlProgramActor {
  sealed trait JobControlLanguage
  final case class SpawnJob(name: String) extends JobControlLanguage
  final case object GracefulShutdown extends JobControlLanguage

  // Predefined cleanup operation
  def cleanup(log: Logger): Unit = log.info("Cleaning up!")

  val mcpa = Behaviors.immutable[JobControlLanguage] { (ctx, msg) ⇒
    msg match {
      case SpawnJob(jobName) ⇒
        ctx.log.info("Spawning job {}!", jobName)
        ctx.spawn(Job.job(jobName), name = jobName)
        Behaviors.same
      case GracefulShutdown ⇒
        ctx.log.info("Initiating graceful shutdown...")
        // perform graceful stop, executing cleanup before final system termination
        // behavior executing cleanup is passed as a parameter to Actor.stopped
        Behaviors.stopped {
          Behaviors.onSignal {
            case (context, PostStop) ⇒
              cleanup(context.system.log)
              Behaviors.same
          }
        }
    }
  }.onSignal {
    case (ctx, PostStop) ⇒
      ctx.log.info("MCPA stopped")
      Behaviors.same
  }
}

object Job {
  import GracefulStopDocSpec.MasterControlProgramActor.JobControlLanguage

  def job(name: String) = Behaviors.onSignal[JobControlLanguage] {
    case (ctx, PostStop) ⇒
      ctx.log.info("Worker {} stopped", name)
      Behaviors.same
  }
}

import MasterControlProgramActor._

val system: ActorSystem[JobControlLanguage] = ActorSystem(mcpa, "B7700")

system ! SpawnJob("a")
system ! SpawnJob("b")

Thread.sleep(100)

// gracefully stop the system
system ! GracefulShutdown

Thread.sleep(100)

Await.result(system.whenTerminated, 3.seconds)

Java

import java.util.concurrent.TimeUnit;

import akka.actor.typed.ActorSystem;
import akka.actor.typed.Behavior;
import akka.actor.typed.PostStop;
import akka.actor.typed.javadsl.Behaviors;


public abstract static class JobControl {
  // no instances of this class, it's only a name space for messages
  // and static methods
  private JobControl() {
  }

  static interface JobControlLanguage {
  }

  public static final class SpawnJob implements JobControlLanguage {
    public final String name;

    public SpawnJob(String name) {
      this.name = name;
    }
  }

  public static final class GracefulShutdown implements JobControlLanguage {

    public GracefulShutdown() {
    }
  }

  public static final Behavior<JobControlLanguage> mcpa = Behaviors.immutable(JobControlLanguage.class)
      .onMessage(SpawnJob.class, (ctx, msg) -> {
        ctx.getSystem().log().info("Spawning job {}!", msg.name);
        ctx.spawn(Job.job(msg.name), msg.name);
        return Behaviors.same();
      })
      .onSignal(PostStop.class, (ctx, signal) -> {
        ctx.getSystem().log().info("Master Control Programme stopped");
        return Behaviors.same();
      })
      .onMessage(GracefulShutdown.class, (ctx, msg) -> {
        ctx.getSystem().log().info("Initiating graceful shutdown...");

        // perform graceful stop, executing cleanup before final system termination
        // behavior executing cleanup is passed as a parameter to Actor.stopped
        return Behaviors.stopped(Behaviors.onSignal((context, PostStop) -> {
          context.getSystem().log().info("Cleanup!");
          return Behaviors.same();
        }));
      })
      .onSignal(PostStop.class, (ctx, signal) -> {
        ctx.getSystem().log().info("Master Control Programme stopped");
        return Behaviors.same();
      })
      .build();
}

public static class Job {
  public static Behavior<JobControl.JobControlLanguage> job(String name) {
    return Behaviors.onSignal((ctx, PostStop) -> {
      ctx.getSystem().log().info("Worker {} stopped", name);
      return Behaviors.same();
    });

  }
}

final ActorSystem<JobControl.JobControlLanguage> system =
    ActorSystem.create(JobControl.mcpa, "B6700");

system.tell(new JobControl.SpawnJob("a"));
system.tell(new JobControl.SpawnJob("b"));

// sleep here to allow time for the new actors to be started
Thread.sleep(100);

system.tell(new JobControl.GracefulShutdown());

system.getWhenTerminated().toCompletableFuture().get(3, TimeUnit.SECONDS);