Interaction Patterns

Dependency

To use Akka Actor Typed, you must add the following dependency in your project:

sbt
libraryDependencies += "com.typesafe.akka" %% "akka-actor-typed" % "2.6-SNAPSHOT"
Maven
<dependency>
  <groupId>com.typesafe.akka</groupId>
  <artifactId>akka-actor-typed_2.12</artifactId>
  <version>2.6-SNAPSHOT</version>
</dependency>
Gradle
dependencies {
  compile group: 'com.typesafe.akka', name: 'akka-actor-typed_2.12', version: '2.6-SNAPSHOT'
}

Introduction

Interacting with an Actor in Akka Typed is done through an ActorRef[T]ActorRef<T> where T is the type of messages the actor accepts, also known as the “protocol”. This ensures that only the right kind of messages can be sent to an actor and also that no one else but the Actor itself can access the Actor instance internals.

Message exchange with Actors follow a few common patterns, let’s go through each one of them.

Fire and Forget

The fundamental way to interact with an actor is through “tell”, which is so common that it has a special symbolic method name: actorRef ! messageactorRef.tell(message). Sending a message with tell can safely be done from any thread.

Tell is asynchronous which means that the method returns right away, when the statement after it is executed there is no guarantee that the message has been processed by the recipient yet. It also means there is no way to know if the message was received, the processing succeeded or failed.

With the given protocol and actor behavior:

Scala
object Printer {

  case class PrintMe(message: String)

  def apply(): Behavior[PrintMe] =
    Behaviors.receive {
      case (context, PrintMe(message)) =>
        context.log.info(message)
        Behaviors.same
    }
}
Java
public class Printer {
  public static class PrintMe {
    public final String message;

    public PrintMe(String message) {
      this.message = message;
    }
  }

  public static Behavior<PrintMe> create() {
    return Behaviors.setup(
        context ->
            Behaviors.receive(PrintMe.class)
                .onMessage(
                    PrintMe.class,
                    printMe -> {
                      context.getLog().info(printMe.message);
                      return Behaviors.same();
                    })
                .build());
  }
}

Fire and forget looks like this:

Scala
val system = ActorSystem(Printer(), "fire-and-forget-sample")

// note how the system is also the top level actor ref
val printer: ActorRef[Printer.PrintMe] = system

// these are all fire and forget
printer ! Printer.PrintMe("message 1")
printer ! Printer.PrintMe("not message 2")
Java
final ActorSystem<Printer.PrintMe> system =
    ActorSystem.create(Printer.create(), "printer-sample-system");

// note that system is also the ActorRef to the guardian actor
final ActorRef<Printer.PrintMe> ref = system;

// these are all fire and forget
ref.tell(new Printer.PrintMe("message 1"));
ref.tell(new Printer.PrintMe("message 2"));

Useful when:

  • It is not critical to be sure that the message was processed
  • There is no way to act on non successful delivery or processing
  • We want to minimize the number of messages created to get higher throughput (sending a response would require creating twice the number of messages)

Problems:

  • If the inflow of messages is higher than the actor can process the inbox will fill up and can in the worst case cause the JVM crash with an OutOfMemoryError
  • If the message gets lost, the sender will not know

Request-Response

Many interactions between actors requires one or more response message being sent back from the receiving actor. A response message can be a result of a query, some form of acknowledgment that the message was received and processed or events that the request subscribed to.

In Akka Typed the recipient of responses has to be encoded as a field in the message itself, which the recipient can then use to send (tell) a response back.

With the following protocol:

Scala
case class Request(query: String, replyTo: ActorRef[Response])
case class Response(result: String)
Java
public static class Request {
  public final String query;
  public final ActorRef<Response> replyTo;

  public Request(String query, ActorRef<Response> replyTo) {
    this.query = query;
    this.replyTo = replyTo;
  }
}

public static class Response {
  public final String result;

  public Response(String result) {
    this.result = result;
  }
}

The sender would use its own ActorRef[Response]ActorRef<Response>, which it can access through ActorContext.selfActorContext.getSelf(), for the respondTo.

Scala
cookieFabric ! CookieFabric.Request("give me cookies", context.self)
Java
cookieFabric.tell(new CookieFabric.Request("give me cookies", context.getSelf()));

On the receiving side the ActorRef[response]ActorRef<Response> can then be used to send one or more responses back:

Scala
def apply(): Behaviors.Receive[Request] =
  Behaviors.receiveMessage[Request] {
    case Request(query, replyTo) =>
      // ... process query ...
      replyTo ! Response(s"Here's the cookies for [$query]!")
      Behaviors.same
  }
Java
// actor behavior
public static Behavior<Request> create() {
  return Behaviors.receive(Request.class)
      .onMessage(Request.class, CookieFabric::onRequest)
      .build();
}

private static Behavior<Request> onRequest(Request request) {
  // ... process request ...
  request.replyTo.tell(new Response("Here's the cookies for " + request.query));
  return Behaviors.same();
}

Useful when:

  • Subscribing to an actor that will send many response messages back

Problems:

  • Actors seldom have a response message from another actor as a part of their protocol (see adapted response)
  • It is hard to detect that a message request was not delivered or processed (see ask)
  • Unless the protocol already includes a way to provide context, for example a request id that is also sent in the response, it is not possible to tie an interaction to some specific context without introducing a new, separate, actor (see ask or per session child actor)

Adapted Response

Most often the sending actor does not, and should not, support receiving the response messages of another actor. In such cases we need to provide an ActorRef of the right type and adapt the response message to a type that the sending actor can handle.

Scala

object Backend { sealed trait Request final case class StartTranslationJob(taskId: Int, site: URI, replyTo: ActorRef[Response]) extends Request sealed trait Response final case class JobStarted(taskId: Int) extends Response final case class JobProgress(taskId: Int, progress: Double) extends Response final case class JobCompleted(taskId: Int, result: URI) extends Response } object Frontend { sealed trait Command final case class Translate(site: URI, replyTo: ActorRef[URI]) extends Command private final case class WrappedBackendResponse(response: Backend.Response) extends Command def apply(backend: ActorRef[Backend.Request]): Behavior[Command] = Behaviors.setup[Command] { context => val backendResponseMapper: ActorRef[Backend.Response] = context.messageAdapter(rsp => WrappedBackendResponse(rsp)) def active(inProgress: Map[Int, ActorRef[URI]], count: Int): Behavior[Command] = { Behaviors.receiveMessage[Command] { case Translate(site, replyTo) => val taskId = count + 1 backend ! Backend.StartTranslationJob(taskId, site, backendResponseMapper) active(inProgress.updated(taskId, replyTo), taskId) case wrapped: WrappedBackendResponse => wrapped.response match { case Backend.JobStarted(taskId) => context.log.info("Started {}", taskId) Behaviors.same case Backend.JobProgress(taskId, progress) => context.log.info2("Progress {}: {}", taskId, progress) Behaviors.same case Backend.JobCompleted(taskId, result) => context.log.info2("Completed {}: {}", taskId, result) inProgress(taskId) ! result active(inProgress - taskId, count) } } } active(inProgress = Map.empty, count = 0) } }
Java

public class Backend { public interface Request {} public static class StartTranslationJob implements Request { public final int taskId; public final URI site; public final ActorRef<Response> replyTo; public StartTranslationJob(int taskId, URI site, ActorRef<Response> replyTo) { this.taskId = taskId; this.site = site; this.replyTo = replyTo; } } public interface Response {} public static class JobStarted implements Response { public final int taskId; public JobStarted(int taskId) { this.taskId = taskId; } } public static class JobProgress implements Response { public final int taskId; public final double progress; public JobProgress(int taskId, double progress) { this.taskId = taskId; this.progress = progress; } } public static class JobCompleted implements Response { public final int taskId; public final URI result; public JobCompleted(int taskId, URI result) { this.taskId = taskId; this.result = result; } } } public class Frontend { public interface Command {} public static class Translate implements Command { public final URI site; public final ActorRef<URI> replyTo; public Translate(URI site, ActorRef<URI> replyTo) { this.site = site; this.replyTo = replyTo; } } private static class WrappedBackendResponse implements Command { final Backend.Response response; public WrappedBackendResponse(Backend.Response response) { this.response = response; } } public static class Translator extends AbstractBehavior<Command> { private final ActorContext<Command> context; private final ActorRef<Backend.Request> backend; private final ActorRef<Backend.Response> backendResponseAdapter; private int taskIdCounter = 0; private Map<Integer, ActorRef<URI>> inProgress = new HashMap<>(); public Translator(ActorContext<Command> context, ActorRef<Backend.Request> backend) { this.context = context; this.backend = backend; this.backendResponseAdapter = context.messageAdapter(Backend.Response.class, WrappedBackendResponse::new); } @Override public Receive<Command> createReceive() { return newReceiveBuilder() .onMessage(Translate.class, this::onTranslate) .onMessage(WrappedBackendResponse.class, this::onWrappedBackendResponse) .build(); } private Behavior<Command> onTranslate(Translate cmd) { taskIdCounter += 1; inProgress.put(taskIdCounter, cmd.replyTo); backend.tell( new Backend.StartTranslationJob(taskIdCounter, cmd.site, backendResponseAdapter)); return this; } private Behavior<Command> onWrappedBackendResponse(WrappedBackendResponse wrapped) { Backend.Response response = wrapped.response; if (response instanceof Backend.JobStarted) { Backend.JobStarted rsp = (Backend.JobStarted) response; context.getLog().info("Started {}", rsp.taskId); } else if (response instanceof Backend.JobProgress) { Backend.JobProgress rsp = (Backend.JobProgress) response; context.getLog().info("Progress {}", rsp.taskId); } else if (response instanceof Backend.JobCompleted) { Backend.JobCompleted rsp = (Backend.JobCompleted) response; context.getLog().info("Completed {}", rsp.taskId); inProgress.get(rsp.taskId).tell(rsp.result); inProgress.remove(rsp.taskId); } else { return Behaviors.unhandled(); } return this; } } }

You can register several message adapters for different message classes. It’s only possible to have one message adapter per message class to make sure that the number of adapters are not growing unbounded if registered repeatedly. That also means that a registered adapter will replace an existing adapter for the same message class.

A message adapter will be used if the message class matches the given class or is a subclass thereof. The registered adapters are tried in reverse order of their registration order, i.e. the last registered first.

A message adapter (and the returned ActorRef) has the same lifecycle as the receiving actor. It’s recommended to register the adapters in a top level Behaviors.setup or constructor of AbstractBehavior but it’s possible to register them later also if needed.

The adapter function is running in the receiving actor and can safely access state of it, but if it throws an exception the actor is stopped.

Useful when:

  • Translating between different actor message protocols
  • Subscribing to an actor that will send many response messages back

Problems:

  • It is hard to detect that a message request was not delivered or processed (see ask)
  • Only one adaption can be made per response message type, if a new one is registered the old one is replaced, for example different target actors can’t have different adaption if they use the same response types, unless some correlation is encoded in the messages
  • Unless the protocol already includes a way to provide context, for example a request id that is also sent in the response, it is not possible to tie an interaction to some specific context without introducing a new, separate, actor

Request-Response with ask between two actors

In an interaction where there is a 1:1 mapping between a request and a response we can use ask on the ActorContext to interact with another actor.

The interaction has two steps, first we need to construct the outgoing message, to do that we need an ActorRef[Response]ActorRef<Response> to put as recipient in the outgoing message. The second step is to transform the successful Response or failure into a message that is part of the protocol of the sending actor.

Scala
object Hal {
  sealed trait Command
  case class OpenThePodBayDoorsPlease(replyTo: ActorRef[Response]) extends Command
  case class Response(message: String)

  def apply(): Behaviors.Receive[Hal.Command] =
    Behaviors.receiveMessage[Command] {
      case OpenThePodBayDoorsPlease(replyTo) =>
        replyTo ! Response("I'm sorry, Dave. I'm afraid I can't do that.")
        Behaviors.same
    }
}

object Dave {

  sealed trait Command
  // this is a part of the protocol that is internal to the actor itself
  private case class AdaptedResponse(message: String) extends Command

  def apply(hal: ActorRef[Hal.Command]): Behavior[Dave.Command] =
    Behaviors.setup[Command] { context =>
      // asking someone requires a timeout, if the timeout hits without response
      // the ask is failed with a TimeoutException
      implicit val timeout: Timeout = 3.seconds

      // Note: The second parameter list takes a function `ActorRef[T] => Message`,
      // as OpenThePodBayDoorsPlease is a case class it has a factory apply method
      // that is what we are passing as the second parameter here it could also be written
      // as `ref => OpenThePodBayDoorsPlease(ref)`
      context.ask(hal, Hal.OpenThePodBayDoorsPlease) {
        case Success(Hal.Response(message)) => AdaptedResponse(message)
        case Failure(_)                     => AdaptedResponse("Request failed")
      }

      // we can also tie in request context into an interaction, it is safe to look at
      // actor internal state from the transformation function, but remember that it may have
      // changed at the time the response arrives and the transformation is done, best is to
      // use immutable state we have closed over like here.
      val requestId = 1
      context.ask(hal, Hal.OpenThePodBayDoorsPlease) {
        case Success(Hal.Response(message)) => AdaptedResponse(s"$requestId: $message")
        case Failure(_)                     => AdaptedResponse(s"$requestId: Request failed")
      }

      Behaviors.receiveMessage {
        // the adapted message ends up being processed like any other
        // message sent to the actor
        case AdaptedResponse(message) =>
          context.log.info("Got response from hal: {}", message)
          Behaviors.same
      }
    }
}
Java
public class Hal extends AbstractBehavior<Hal.Command> {

  public interface Command {}

  public static final class OpenThePodBayDoorsPlease implements Command {
    public final ActorRef<HalResponse> respondTo;

    public OpenThePodBayDoorsPlease(ActorRef<HalResponse> respondTo) {
      this.respondTo = respondTo;
    }
  }

  public static final class HalResponse {
    public final String message;

    public HalResponse(String message) {
      this.message = message;
    }
  }

  @Override
  public Receive<Command> createReceive() {
    return newReceiveBuilder()
        .onMessage(OpenThePodBayDoorsPlease.class, this::onOpenThePodBayDoorsPlease)
        .build();
  }

  private Behavior<Command> onOpenThePodBayDoorsPlease(OpenThePodBayDoorsPlease message) {
    message.respondTo.tell(new HalResponse("I'm sorry, Dave. I'm afraid I can't do that."));
    return this;
  }
}

public class Dave extends AbstractBehavior<Dave.Command> {

  public interface Command {}

  // this is a part of the protocol that is internal to the actor itself
  private static final class AdaptedResponse implements Command {
    public final String message;

    public AdaptedResponse(String message) {
      this.message = message;
    }
  }

  public static Behavior<Command> create(ActorRef<Hal.Command> hal) {
    return Behaviors.setup(context -> new Dave(context, hal));
  }

  private final ActorContext<Command> context;

  private Dave(ActorContext<Command> context, ActorRef<Hal.Command> hal) {
    this.context = context;

    // asking someone requires a timeout, if the timeout hits without response
    // the ask is failed with a TimeoutException
    final Duration timeout = Duration.ofSeconds(3);

    context.ask(
        Hal.HalResponse.class,
        hal,
        timeout,
        // construct the outgoing message
        (ActorRef<Hal.HalResponse> ref) -> new Hal.OpenThePodBayDoorsPlease(ref),
        // adapt the response (or failure to respond)
        (response, throwable) -> {
          if (response != null) {
            return new AdaptedResponse(response.message);
          } else {
            return new AdaptedResponse("Request failed");
          }
        });

    // we can also tie in request context into an interaction, it is safe to look at
    // actor internal state from the transformation function, but remember that it may have
    // changed at the time the response arrives and the transformation is done, best is to
    // use immutable state we have closed over like here.
    final int requestId = 1;
    context.ask(
        Hal.HalResponse.class,
        hal,
        timeout,
        // construct the outgoing message
        (ActorRef<Hal.HalResponse> ref) -> new Hal.OpenThePodBayDoorsPlease(ref),
        // adapt the response (or failure to respond)
        (response, throwable) -> {
          if (response != null) {
            return new AdaptedResponse(requestId + ": " + response.message);
          } else {
            return new AdaptedResponse(requestId + ": Request failed");
          }
        });
  }

  @Override
  public Receive<Command> createReceive() {
    return newReceiveBuilder()
        // the adapted message ends up being processed like any other
        // message sent to the actor
        .onMessage(AdaptedResponse.class, this::onAdaptedResponse)
        .build();
  }

  private Behavior<Command> onAdaptedResponse(AdaptedResponse response) {
    context.getLog().info("Got response from HAL: {}", response.message);
    return this;
  }
}

The response adapting function is running in the receiving actor and can safely access state of it, but if it throws an exception the actor is stopped.

Useful when:

  • Single response queries
  • An actor needs to know that the message was processed before continuing
  • To allow an actor to resend if a timely response is not produced
  • To keep track of outstanding requests and not overwhelm a recipient with messages (“backpressure”)
  • Context should be attached to the interaction but the protocol does not support that (request id, what query the response was for)

Problems:

  • There can only be a single response to one ask (see per session child Actor)
  • When ask times out, the receiving actor does not know and may still process it to completion, or even start processing it after the fact
  • Finding a good value for the timeout, especially when ask is triggers chained asks in the receiving actor. You want a short timeout to be responsive and answer back to the requester, but at the same time you do not want to have many false positives

Request-Response with ask from outside an Actor

Some times you need to interact with actors from outside of the actor system, this can be done with fire-and-forget as described above or through another version of ask that returns a Future[Response]CompletionStage<Response> that is either completed with a successful response or failed with a TimeoutException if there was no response within the specified timeout.

To do this we use ActorRef.ask (or the symbolic ActorRef.?) implicitly provided by akka.actor.typed.scaladsl.AskPatternakka.actor.typed.javadsl.AskPattern.ask to send a message to an actor and get a Future[Response]CompletionState[Response] back.

Scala
object CookieFabric {
  sealed trait Command {}
  case class GiveMeCookies(replyTo: ActorRef[Cookies]) extends Command
  case class Cookies(count: Int)

  def apply(): Behaviors.Receive[CookieFabric.GiveMeCookies] =
    Behaviors.receiveMessage { message =>
      message.replyTo ! Cookies(5)
      Behaviors.same
    }
}

import akka.actor.typed.scaladsl.AskPattern._
import akka.util.Timeout

// asking someone requires a timeout and a scheduler, if the timeout hits without response
// the ask is failed with a TimeoutException
implicit val timeout: Timeout = 3.seconds
implicit val scheduler = system.scheduler

val result: Future[CookieFabric.Cookies] = cookieFabric.ask(ref => CookieFabric.GiveMeCookies(ref))

// the response callback will be executed on this execution context
implicit val ec = system.executionContext

result.onComplete {
  case Success(cookies) => println(s"Yay, cookies! $cookies")
  case Failure(ex)      => println(s"Boo! didn't get cookies: ${ex.getMessage}")
}
Java
public class CookieFabric {

  interface Command {}

  public static class GiveMeCookies implements Command {
    public final ActorRef<Cookies> cookies;

    public GiveMeCookies(ActorRef<Cookies> cookies) {
      this.cookies = cookies;
    }
  }

  public static class Cookies {
    public final int count;

    public Cookies(int count) {
      this.count = count;
    }
  }
}

  public void askAndPrint(
      ActorSystem<Void> system, ActorRef<CookieFabric.Command> cookieFabric) {
    CompletionStage<CookieFabric.Cookies> result =
        AskPattern.ask(
            cookieFabric,
            CookieFabric.GiveMeCookies::new,
            // asking someone requires a timeout and a scheduler, if the timeout hits without
            // response
            // the ask is failed with a TimeoutException
            Duration.ofSeconds(3),
            system.scheduler());

    result.whenComplete(
        (cookies, failure) -> {
          if (cookies != null) System.out.println("Yay, cookies!");
          else System.out.println("Boo! didn't get cookies in time.");
        });
  }

Useful when:

  • Querying an actor from outside of the actor system

Problems:

  • It is easy to accidentally close over and unsafely mutable state with the callbacks on the returned FutureCompletionStage as those will be executed on a different thread
  • There can only be a single response to one ask (see per session child Actor)
  • When ask times out, the receiving actor does not know and may still process it to completion, or even start processing it after the fact

Per session child Actor

In some cases a complete response to a request can only be created and sent back after collecting multiple answers from other actors. For these kinds of interaction it can be good to delegate the work to a per “session” child actor. The child could also contain arbitrary logic to implement retrying, failing on timeout, tail chopping, progress inspection etc.

Note that this in fact essentially how ask is implemented, if all you need is a single response with a timeout it is better to use ask.

The child is created with the context it needs to do the work, including an ActorRef that it can respond to. When the complete result is there the child responds with the result and stops itself.

As the protocol of the session actor is not a public API but rather an implementation detail of the parent actor, it may not always make sense to have an explicit protocol and adapt the messages of the actors that the session actor interacts with. For this use case it is possible to express that the actor can receive any message (AnyObject).

Scala
// dummy data types just for this sample
case class Keys()
case class Wallet()


object Home {
  sealed trait Command
  case class LeaveHome(who: String, replyTo: ActorRef[ReadyToLeaveHome]) extends Command
  case class ReadyToLeaveHome(who: String, keys: Keys, wallet: Wallet)

  def apply(): Behavior[Command] = {
    Behaviors.setup[Command] { context =>
      val keyCabinet: ActorRef[KeyCabinet.GetKeys] = context.spawn(KeyCabinet(), "key-cabinet")
      val drawer: ActorRef[Drawer.GetWallet] = context.spawn(Drawer(), "drawer")

      Behaviors.receiveMessage[Command] {
        case LeaveHome(who, replyTo) =>
          context.spawn(prepareToLeaveHome(who, replyTo, keyCabinet, drawer), s"leaving-$who")
          Behaviors.same
      }
    }
  }

  // per session actor behavior
  def prepareToLeaveHome(
      whoIsLeaving: String,
      replyTo: ActorRef[ReadyToLeaveHome],
      keyCabinet: ActorRef[KeyCabinet.GetKeys],
      drawer: ActorRef[Drawer.GetWallet]): Behavior[NotUsed] = {
    // we don't _really_ care about the actor protocol here as nobody will send us
    // messages except for responses to our queries, so we just accept any kind of message
    // but narrow that to more limited types when we interact
    Behaviors
      .setup[AnyRef] { context =>
        var wallet: Option[Wallet] = None
        var keys: Option[Keys] = None

        // we narrow the ActorRef type to any subtype of the actual type we accept
        keyCabinet ! KeyCabinet.GetKeys(whoIsLeaving, context.self.narrow[Keys])
        drawer ! Drawer.GetWallet(whoIsLeaving, context.self.narrow[Wallet])

        def nextBehavior(): Behavior[AnyRef] =
          (keys, wallet) match {
            case (Some(w), Some(k)) =>
              // we got both, "session" is completed!
              replyTo ! ReadyToLeaveHome(whoIsLeaving, w, k)
              Behaviors.stopped

            case _ =>
              Behaviors.same
          }

        Behaviors.receiveMessage {
          case w: Wallet =>
            wallet = Some(w)
            nextBehavior()
          case k: Keys =>
            keys = Some(k)
            nextBehavior()
          case _ =>
            Behaviors.unhandled
        }
      }
      .narrow[NotUsed] // we don't let anyone else know we accept anything
  }
}
Java
// dummy data types just for this sample
public class Keys {}

public class Wallet {}

public class KeyCabinet {
  public static class GetKeys {
    public final String whoseKeys;
    public final ActorRef<Keys> replyTo;

    public GetKeys(String whoseKeys, ActorRef<Keys> respondTo) {
      this.whoseKeys = whoseKeys;
      this.replyTo = respondTo;
    }
  }

  public static Behavior<GetKeys> create() {
    return Behaviors.receiveMessage(KeyCabinet::onGetKeys);
  }

  private static Behavior<GetKeys> onGetKeys(GetKeys message) {
    message.replyTo.tell(new Keys());
    return Behaviors.same();
  }
}

public class Drawer {

  public static class GetWallet {
    public final String whoseWallet;
    public final ActorRef<Wallet> replyTo;

    public GetWallet(String whoseWallet, ActorRef<Wallet> replyTo) {
      this.whoseWallet = whoseWallet;
      this.replyTo = replyTo;
    }
  }

  public static Behavior<GetWallet> create() {
    return Behaviors.receiveMessage(Drawer::onGetWallet);
  }

  private static Behavior<GetWallet> onGetWallet(GetWallet message) {
    message.replyTo.tell(new Wallet());
    return Behaviors.same();
  }
}

public class Home {

  public interface Command {}

  public static class LeaveHome implements Command {
    public final String who;
    public final ActorRef<ReadyToLeaveHome> respondTo;

    public LeaveHome(String who, ActorRef<ReadyToLeaveHome> respondTo) {
      this.who = who;
      this.respondTo = respondTo;
    }
  }

  public static class ReadyToLeaveHome {
    public final String who;
    public final Keys keys;
    public final Wallet wallet;

    public ReadyToLeaveHome(String who, Keys keys, Wallet wallet) {
      this.who = who;
      this.keys = keys;
      this.wallet = wallet;
    }
  }

  private final ActorContext<Command> context;

  private final ActorRef<KeyCabinet.GetKeys> keyCabinet;
  private final ActorRef<Drawer.GetWallet> drawer;

  private Home(ActorContext<Command> context) {
    this.context = context;
    this.keyCabinet = context.spawn(KeyCabinet.create(), "key-cabinet");
    this.drawer = context.spawn(Drawer.create(), "drawer");
  }

  private Behavior<Command> behavior() {
    return Behaviors.receive(Command.class)
        .onMessage(LeaveHome.class, this::onLeaveHome)
        .build();
  }

  private Behavior<Command> onLeaveHome(LeaveHome message) {
    context.spawn(
        new PrepareToLeaveHome(message.who, message.respondTo, keyCabinet, drawer),
        "leaving" + message.who);
    return Behaviors.same();
  }

  // actor behavior
  public static Behavior<Command> create() {
    return Behaviors.setup(context -> new Home(context).behavior());
  }
}

// per session actor behavior
class PrepareToLeaveHome extends AbstractBehavior<Object> {
  private final String whoIsLeaving;
  private final ActorRef<Home.ReadyToLeaveHome> replyTo;
  private final ActorRef<KeyCabinet.GetKeys> keyCabinet;
  private final ActorRef<Drawer.GetWallet> drawer;
  private Optional<Wallet> wallet = Optional.empty();
  private Optional<Keys> keys = Optional.empty();

  PrepareToLeaveHome(
      String whoIsLeaving,
      ActorRef<Home.ReadyToLeaveHome> replyTo,
      ActorRef<KeyCabinet.GetKeys> keyCabinet,
      ActorRef<Drawer.GetWallet> drawer) {
    this.whoIsLeaving = whoIsLeaving;
    this.replyTo = replyTo;
    this.keyCabinet = keyCabinet;
    this.drawer = drawer;
  }

  @Override
  public Receive<Object> createReceive() {
    return newReceiveBuilder()
        .onMessage(Wallet.class, this::onWallet)
        .onMessage(Keys.class, this::onKeys)
        .build();
  }

  private Behavior<Object> onWallet(Wallet wallet) {
    this.wallet = Optional.of(wallet);
    return completeOrContinue();
  }

  private Behavior<Object> onKeys(Keys keys) {
    this.keys = Optional.of(keys);
    return completeOrContinue();
  }

  private Behavior<Object> completeOrContinue() {
    if (wallet.isPresent() && keys.isPresent()) {
      replyTo.tell(new Home.ReadyToLeaveHome(whoIsLeaving, keys.get(), wallet.get()));
      return Behaviors.stopped();
    } else {
      return this;
    }
  }
}

In an actual session child you would likely want to include some form of timeout as well (see scheduling messages to self).

Useful when:

  • A single incoming request should result in multiple interactions with other actors before a result can be built, for example aggregation of several results
  • You need to handle acknowledgement and retry messages for at-least-once delivery

Problems:

  • Children have life cycles that must be managed to not create a resource leak, it can be easy to miss a scenario where the session actor is not stopped
  • It increases complexity, since each such child can execute concurrently with other children and the parent

Scheduling messages to self

The following example demonstrates how to use timers to schedule messages to an actor.

The Buncher actor buffers a burst of incoming messages and delivers them as a batch after a timeout or when the number of batched messages exceeds a maximum size.

Scala
object Buncher {

  sealed trait Command
  final case class ExcitingMessage(message: String) extends Command
  final case class Batch(messages: Vector[Command])
  private case object Timeout extends Command
  private case object TimerKey

  def apply(target: ActorRef[Batch], after: FiniteDuration, maxSize: Int): Behavior[Command] = {
    Behaviors.withTimers(timers => new Buncher(timers, target, after, maxSize).idle())
  }
}

class Buncher(
    timers: TimerScheduler[Buncher.Command],
    target: ActorRef[Buncher.Batch],
    after: FiniteDuration,
    maxSize: Int) {
  import Buncher._

  private def idle(): Behavior[Command] = {
    Behaviors.receiveMessage[Command] { message =>
      timers.startSingleTimer(TimerKey, Timeout, after)
      active(Vector(message))
    }
  }

  def active(buffer: Vector[Command]): Behavior[Command] = {
    Behaviors.receiveMessage[Command] {
      case Timeout =>
        target ! Batch(buffer)
        idle()
      case m =>
        val newBuffer = buffer :+ m
        if (newBuffer.size == maxSize) {
          timers.cancel(TimerKey)
          target ! Batch(newBuffer)
          idle()
        } else
          active(newBuffer)
    }
  }
}
Java
public class Buncher {

  public interface Command {}

  public static final class Batch {
    private final List<Command> messages;

    public Batch(List<Command> messages) {
      this.messages = Collections.unmodifiableList(messages);
    }

    public List<Command> getMessages() {
      return messages;
    }
  }

  public static final class ExcitingMessage implements Command {
    public final String message;

    public ExcitingMessage(String message) {
      this.message = message;
    }
  }

  private static final Object TIMER_KEY = new Object();

  private enum Timeout implements Command {
    INSTANCE
  }

  public static Behavior<Command> create(ActorRef<Batch> target, Duration after, int maxSize) {
    return Behaviors.withTimers(timers -> new Buncher(timers, target, after, maxSize).idle());
  }

  private final TimerScheduler<Command> timers;
  private final ActorRef<Batch> target;
  private final Duration after;
  private final int maxSize;

  private Buncher(
      TimerScheduler<Command> timers, ActorRef<Batch> target, Duration after, int maxSize) {
    this.timers = timers;
    this.target = target;
    this.after = after;
    this.maxSize = maxSize;
  }

  private Behavior<Command> idle() {
    return Behaviors.receive(Command.class)
        .onMessage(Command.class, this::onIdleCommand)
        .build();
  }

  private Behavior<Command> onIdleCommand(Command message) {
    timers.startSingleTimer(TIMER_KEY, Timeout.INSTANCE, after);
    return new Active(message);
  }

  private class Active extends AbstractBehavior<Command> {
    private final List<Command> buffer = new ArrayList<>();

    Active(Command firstCommand) {
      buffer.add(firstCommand);
    }

    @Override
    public Receive<Command> createReceive() {
      return newReceiveBuilder()
          .onMessage(Timeout.class, message -> onTimeout())
          .onMessage(Command.class, this::onCommand)
          .build();
    }

    private Behavior<Command> onTimeout() {
      target.tell(new Batch(buffer));
      return idle(); // switch to idle
    }

    private Behavior<Command> onCommand(Command message) {
      buffer.add(message);
      if (buffer.size() == maxSize) {
        timers.cancel(TIMER_KEY);
        target.tell(new Batch(buffer));
        return idle(); // switch to idle
      } else {
        return this; // stay Active
      }
    }
  }
}

There are a few things worth noting here:

  • To get access to the timers you start with Behaviors.withTimers that will pass a TimerScheduler instance to the function. This can be used with any type of Behavior, including receive, receiveMessage, but also setup or any other behavior.
  • Each timer has a key and if a new timer with same key is started the previous is cancelled and it’s guaranteed that a message from the previous timer is not received, even though it might already be enqueued in the mailbox when the new timer is started.
  • Both periodic and single message timers are supported.
  • The TimerScheduler is mutable in itself, because it performs and manages the side effects of registering the scheduled tasks.
  • The TimerScheduler is bound to the lifecycle of the actor that owns it and it’s cancelled automatically when the actor is stopped.
  • Behaviors.withTimers can also be used inside Behaviors.supervise and it will automatically cancel the started timers correctly when the actor is restarted, so that the new incarnation will not receive scheduled messages from previous incarnation.

Schedule periodically

Scheduling of recurring messages can have two different characteristics:

  • fixed-delay - The delay between sending subsequent messages will always be (at least) the given delay. Use startTimerWithFixedDelay.
  • fixed-rate - The frequency of execution over time will meet the given interval. Use startTimerAtFixedRate.

If you are uncertain of which one to use you should pick startTimerWithFixedDelay.

When using fixed-delay it will not compensate the delay between messages if the scheduling is delayed longer than specified for some reason. The delay between sending subsequent messages will always be (at least) the given delay. In the long run, the frequency of messages will generally be slightly lower than the reciprocal of the specified delay.

Fixed-delay execution is appropriate for recurring activities that require “smoothness.” In other words, it is appropriate for activities where it is more important to keep the frequency accurate in the short run than in the long run.

When using fixed-rate it will compensate the delay for a subsequent task if the previous messages were delayed too long. In such cases, the actual sending interval will differ from the interval passed to the scheduleAtFixedRate method.

If the tasks are delayed longer than the interval, the subsequent message will be sent immediately after the prior one. This also has the consequence that after long garbage collection pauses or other reasons when the JVM was suspended all “missed” tasks will execute when the process wakes up again. For example, scheduleAtFixedRate with an interval of 1 second and the process is suspended for 30 seconds will result in 30 messages being sent in rapid succession to catch up. In the long run, the frequency of execution will be exactly the reciprocal of the specified interval.

Fixed-rate execution is appropriate for recurring activities that are sensitive to absolute time or where the total time to perform a fixed number of executions is important, such as a countdown timer that ticks once every second for ten seconds.

Warning

scheduleAtFixedRate can result in bursts of scheduled messages after long garbage collection pauses, which may in worst case cause undesired load on the system. scheduleWithFixedDelay is often preferred.

Responding to a sharded actor

The normal pattern for expecting a reply is to include an ActorRefActorRef in the message, typically a message adapter. This can be used for a sharded actor but if ctx.selfctx.getSelf() is sent and the sharded actor is moved or passivated then the reply will sent to dead letters.

An alternative is to send the entityId in the message and have the reply sent via sharding:

Scala
// a sharded actor that needs counter updates
object CounterConsumer {
  sealed trait Command
  final case class NewCount(count: Long) extends Command
  val TypeKey: EntityTypeKey[Command] = EntityTypeKey[Command]("example-sharded-response")
}

// a sharded counter that sends responses to another sharded actor
object Counter {
  trait Command
  case object Increment extends Command
  final case class GetValue(replyToEntityId: String) extends Command
  val TypeKey: EntityTypeKey[Command] = EntityTypeKey[Command]("example-sharded-counter")

  private def apply(): Behavior[Command] =
    Behaviors.setup { context =>
      counter(ClusterSharding(context.system), 0)
    }

  private def counter(sharding: ClusterSharding, value: Long): Behavior[Command] =
    Behaviors.receiveMessage {
      case Increment =>
        counter(sharding, value + 1)
      case GetValue(replyToEntityId) =>
        val replyToEntityRef = sharding.entityRefFor(CounterConsumer.TypeKey, replyToEntityId)
        replyToEntityRef ! CounterConsumer.NewCount(value)
        Behaviors.same
    }

}
Java
// a sharded actor that needs counter updates
public class CounterConsumer {
  public static EntityTypeKey<Command> typeKey =
      EntityTypeKey.create(Command.class, "example-sharded-response");

  public interface Command {}

  public static class NewCount implements Command {
    public final long value;

    public NewCount(long value) {
      this.value = value;
    }
  }
}

// a sharded counter that sends responses to another sharded actor
public class Counter extends AbstractBehavior<Counter.Command> {
  public static EntityTypeKey<Command> typeKey =
      EntityTypeKey.create(Command.class, "example-sharded-counter");

  public interface Command {}

  public enum Increment implements Command {
    INSTANCE
  }

  public static class GetValue implements Command {
    public final String replyToEntityId;

    public GetValue(String replyToEntityId) {
      this.replyToEntityId = replyToEntityId;
    }
  }

  public static Behavior<Command> create() {
    return Behaviors.setup(Counter::new);
  }

  private final ClusterSharding sharding;
  private int value = 0;

  private Counter(ActorContext<Command> context) {
    this.sharding = ClusterSharding.get(context.getSystem());
  }

  @Override
  public Receive<Command> createReceive() {
    return newReceiveBuilder()
        .onMessage(Increment.class, msg -> onIncrement())
        .onMessage(GetValue.class, this::onGetValue)
        .build();
  }

  private Behavior<Command> onIncrement() {
    value++;
    return this;
  }

  private Behavior<Command> onGetValue(GetValue msg) {
    EntityRef<CounterConsumer.Command> entityRef =
        sharding.entityRefFor(CounterConsumer.typeKey, msg.replyToEntityId);
    entityRef.tell(new CounterConsumer.NewCount(value));
    return this;
  }
}

A disadvantage is that a message adapter can’t be used so the response has to be in the protocol of the actor being responded to. Additionally the EntityTypeKey could be included in the message if it is not known statically.

Found an error in this documentation? The source code for this page can be found here. Please feel free to edit and contribute a pull request.