Interaction Patterns

<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencyManagement>
  <dependencies>
    <dependency>
      <groupId>org.apache.pekko</groupId>
      <artifactId>pekko-bom_${scala.binary.version}</artifactId>
      <version>1.1.3</version>
      <type>pom</type>
      <scope>import</scope>
    </dependency>
  </dependencies>
</dependencyManagement>
<dependencies>
  <dependency>
    <groupId>org.apache.pekko</groupId>
    <artifactId>pekko-actor-typed_${scala.binary.version}</artifactId>
  </dependency>
</dependencies>

def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation platform("org.apache.pekko:pekko-bom_${versions.ScalaBinary}:1.1.3")

  implementation "org.apache.pekko:pekko-actor-typed_${versions.ScalaBinary}"
}

copysourcepublic 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());
  }
}

copysourcefinal 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"));

copysourcepublic 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;
  }
}

copysourcecookieFabric.tell(new CookieFabric.Request("give me cookies", context.getSelf()));

copysource// 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 are the cookies for " + request.query));
  return Behaviors.same();
}

copysource
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 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) {
      super(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;
        getContext().getLog().info("Started {}", rsp.taskId);
      } else if (response instanceof Backend.JobProgress) {
        Backend.JobProgress rsp = (Backend.JobProgress) response;
        getContext().getLog().info("Progress {}", rsp.taskId);
      } else if (response instanceof Backend.JobCompleted) {
        Backend.JobCompleted rsp = (Backend.JobCompleted) response;
        getContext().getLog().info("Completed {}", rsp.taskId);
        inProgress.get(rsp.taskId).tell(rsp.result);
        inProgress.remove(rsp.taskId);
      } else {
        return Behaviors.unhandled();
      }

      return this;
    }
  }
}

copysourcepublic class Hal extends AbstractBehavior<Hal.Command> {

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

  private Hal(ActorContext<Command> context) {
    super(context);
  }

  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 Dave(ActorContext<Command> context, ActorRef<Hal.Command> hal) {
    super(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) {
    getContext().getLog().info("Got response from HAL: {}", response.message);
    return this;
  }
}

copysourcepublic class CookieFabric extends AbstractBehavior<CookieFabric.Command> {

  interface Command {}

  public static class GiveMeCookies implements Command {
    public final int count;
    public final ActorRef<Reply> replyTo;

    public GiveMeCookies(int count, ActorRef<Reply> replyTo) {
      this.count = count;
      this.replyTo = replyTo;
    }
  }

  interface Reply {}

  public static class Cookies implements Reply {
    public final int count;

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

  public static class InvalidRequest implements Reply {
    public final String reason;

    public InvalidRequest(String reason) {
      this.reason = reason;
    }
  }

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

  private CookieFabric(ActorContext<Command> context) {
    super(context);
  }

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

  private Behavior<Command> onGiveMeCookies(GiveMeCookies request) {
    if (request.count >= 5) request.replyTo.tell(new InvalidRequest("Too many cookies."));
    else request.replyTo.tell(new Cookies(request.count));

    return this;
  }
}

  public void askAndPrint(
      ActorSystem<Void> system, ActorRef<CookieFabric.Command> cookieFabric) {
    CompletionStage<CookieFabric.Reply> result =
        AskPattern.ask(
            cookieFabric,
            replyTo -> new CookieFabric.GiveMeCookies(3, replyTo),
            // 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(
        (reply, failure) -> {
          if (reply instanceof CookieFabric.Cookies)
            System.out.println("Yay, " + ((CookieFabric.Cookies) reply).count + " cookies!");
          else if (reply instanceof CookieFabric.InvalidRequest)
            System.out.println(
                "No cookies for me. " + ((CookieFabric.InvalidRequest) reply).reason);
          else System.out.println("Boo! didn't get cookies in time. " + failure);
        });
  }

copysourceCompletionStage<CookieFabric.Reply> result =
    AskPattern.ask(
        cookieFabric,
        replyTo -> new CookieFabric.GiveMeCookies(3, replyTo),
        Duration.ofSeconds(3),
        system.scheduler());

CompletionStage<CookieFabric.Cookies> cookies =
    result.thenCompose(
        (CookieFabric.Reply reply) -> {
          if (reply instanceof CookieFabric.Cookies) {
            return CompletableFuture.completedFuture((CookieFabric.Cookies) reply);
          } else if (reply instanceof CookieFabric.InvalidRequest) {
            CompletableFuture<CookieFabric.Cookies> failed = new CompletableFuture<>();
            failed.completeExceptionally(
                new IllegalArgumentException(((CookieFabric.InvalidRequest) reply).reason));
            return failed;
          } else {
            throw new IllegalStateException("Unexpected reply: " + reply.getClass());
          }
        });

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

copysourceimport org.apache.pekko.pattern.StatusReply;

public class Hal extends AbstractBehavior<Hal.Command> {

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

  private Hal(ActorContext<Hal.Command> context) {
    super(context);
  }

  public interface Command {}

  public static final class OpenThePodBayDoorsPlease implements Hal.Command {
    public final ActorRef<StatusReply<String>> respondTo;

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

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

  private Behavior<Hal.Command> onOpenThePodBayDoorsPlease(
      Hal.OpenThePodBayDoorsPlease message) {
    message.respondTo.tell(StatusReply.error("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 Dave.Command {
    public final String message;

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

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

  private Dave(ActorContext<Dave.Command> context, ActorRef<Hal.Command> hal) {
    super(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.askWithStatus(
        String.class,
        hal,
        timeout,
        // construct the outgoing message
        (ActorRef<StatusReply<String>> ref) -> new Hal.OpenThePodBayDoorsPlease(ref),
        // adapt the response (or failure to respond)
        (response, throwable) -> {
          if (response != null) {
            // a ReponseWithStatus.success(m) is unwrapped and passed as response
            return new Dave.AdaptedResponse(response);
          } else {
            // a ResponseWithStatus.error will end up as a StatusReply.ErrorMessage()
            // exception here
            return new Dave.AdaptedResponse("Request failed: " + throwable.getMessage());
          }
        });
  }

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

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

copysourcepublic class CookieFabric extends AbstractBehavior<CookieFabric.Command> {

  interface Command {}

  public static class GiveMeCookies implements CookieFabric.Command {
    public final int count;
    public final ActorRef<StatusReply<CookieFabric.Cookies>> replyTo;

    public GiveMeCookies(int count, ActorRef<StatusReply<CookieFabric.Cookies>> replyTo) {
      this.count = count;
      this.replyTo = replyTo;
    }
  }

  public static class Cookies {
    public final int count;

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

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

  private CookieFabric(ActorContext<CookieFabric.Command> context) {
    super(context);
  }

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

  private Behavior<CookieFabric.Command> onGiveMeCookies(CookieFabric.GiveMeCookies request) {
    if (request.count >= 5) request.replyTo.tell(StatusReply.error("Too many cookies."));
    else request.replyTo.tell(StatusReply.success(new CookieFabric.Cookies(request.count)));

    return this;
  }
}

  public void askAndPrint(
      ActorSystem<Void> system, ActorRef<CookieFabric.Command> cookieFabric) {
    CompletionStage<CookieFabric.Cookies> result =
        AskPattern.askWithStatus(
            cookieFabric,
            replyTo -> new CookieFabric.GiveMeCookies(3, replyTo),
            // 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(
        (reply, failure) -> {
          if (reply != null) System.out.println("Yay, " + reply.count + " cookies!");
          else if (failure instanceof StatusReply.ErrorMessage)
            System.out.println("No cookies for me. " + failure.getMessage());
          else System.out.println("Boo! didn't get cookies in time. " + failure);
        });
  }

copysourceCompletionStage<CookieFabric.Cookies> cookies =
    AskPattern.askWithStatus(
        cookieFabric,
        replyTo -> new CookieFabric.GiveMeCookies(3, replyTo),
        Duration.ofSeconds(3),
        system.scheduler());

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

copysourcecookieFabric.tell(
    new CookieFabric.Request("don't send cookies back", context.getSystem().ignoreRef()));

copysourcepublic interface CustomerDataAccess {
  CompletionStage<Done> update(Customer customer);
}

public class Customer {
  public final String id;
  public final long version;
  public final String name;
  public final String address;

  public Customer(String id, long version, String name, String address) {
    this.id = id;
    this.version = version;
    this.name = name;
    this.address = address;
  }
}

public class CustomerRepository extends AbstractBehavior<CustomerRepository.Command> {

  private static final int MAX_OPERATIONS_IN_PROGRESS = 10;

  interface Command {}

  public static class Update implements Command {
    public final Customer customer;
    public final ActorRef<OperationResult> replyTo;

    public Update(Customer customer, ActorRef<OperationResult> replyTo) {
      this.customer = customer;
      this.replyTo = replyTo;
    }
  }

  interface OperationResult {}

  public static class UpdateSuccess implements OperationResult {
    public final String id;

    public UpdateSuccess(String id) {
      this.id = id;
    }
  }

  public static class UpdateFailure implements OperationResult {
    public final String id;
    public final String reason;

    public UpdateFailure(String id, String reason) {
      this.id = id;
      this.reason = reason;
    }
  }

  private static class WrappedUpdateResult implements Command {
    public final OperationResult result;
    public final ActorRef<OperationResult> replyTo;

    private WrappedUpdateResult(OperationResult result, ActorRef<OperationResult> replyTo) {
      this.result = result;
      this.replyTo = replyTo;
    }
  }

  public static Behavior<Command> create(CustomerDataAccess dataAccess) {
    return Behaviors.setup(context -> new CustomerRepository(context, dataAccess));
  }

  private final CustomerDataAccess dataAccess;
  private int operationsInProgress = 0;

  private CustomerRepository(ActorContext<Command> context, CustomerDataAccess dataAccess) {
    super(context);
    this.dataAccess = dataAccess;
  }

  @Override
  public Receive<Command> createReceive() {
    return newReceiveBuilder()
        .onMessage(Update.class, this::onUpdate)
        .onMessage(WrappedUpdateResult.class, this::onUpdateResult)
        .build();
  }

  private Behavior<Command> onUpdate(Update command) {
    if (operationsInProgress == MAX_OPERATIONS_IN_PROGRESS) {
      command.replyTo.tell(
          new UpdateFailure(
              command.customer.id,
              "Max " + MAX_OPERATIONS_IN_PROGRESS + " concurrent operations supported"));
    } else {
      // increase operationsInProgress counter
      operationsInProgress++;
      CompletionStage<Done> futureResult = dataAccess.update(command.customer);
      getContext()
          .pipeToSelf(
              futureResult,
              (ok, exc) -> {
                if (exc == null)
                  return new WrappedUpdateResult(
                      new UpdateSuccess(command.customer.id), command.replyTo);
                else
                  return new WrappedUpdateResult(
                      new UpdateFailure(command.customer.id, exc.getMessage()),
                      command.replyTo);
              });
    }
    return this;
  }

  private Behavior<Command> onUpdateResult(WrappedUpdateResult wrapped) {
    // decrease operationsInProgress counter
    operationsInProgress--;
    // send result to original requestor
    wrapped.replyTo.tell(wrapped.result);
    return this;
  }
}