Integrating with Akka Typed & Cluster Sharding

Introduction

Akka is an open-source and free toolkit and runtime for designing highly concurrent, fault-tolerant, and distributed applications on the Java Virtual Machine. The Akka framework gives good abstraction for asynchronous, concurrent, and distributed programming, like Actors, Futures, and streams.

Akka contributes multiple programming models for concurrency, but it asserts actor-based concurrency. Language binding exists for both Scala and Java. Akka is written in the Scala language, and as of scala 2.10, the actors in the scala library are belittled in favor of Akka.

Integrating with Akka Typed

Akka 2.6 added the new typed Actor API(“Akka Typed”) as stable. Now, the typed API is the main API for Akka officially. In typed API, every actor must declare which message type it can handle, and the type system carries out that only messages of this type can be sent to the actor. As Play does not fully adopt Akka Typed, we already give some APIs to integrate it in Play apps.

Akka Actor Types styles

Akka’s Actor Typed API consists of two styles:

A functional programming style based on defining an actor's Behaviour s with values; and
An object-oriented style is based on defining an actor's Behaviour s with subclasses.

Below is the example of a simple actor that says “hello back”:

📘Scala FP

import akka.actor.typed.ActorRef
import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl.Behaviors


object HelloActor {
  final case class SayHello(
      name: String,
      replyTo: ActorRef[String],
  )


  def create(): Behavior[SayHello] = {
    Behaviors.receiveMessage[SayHello] {
      case SayHello(name, replyTo) =>
        replyTo ! s"Hello, $name"
        Behaviors.same
    }
  }
}

📘 Scala OO

import akka.actor.typed.ActorRef
import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl.AbstractBehavior
import akka.actor.typed.scaladsl.ActorContext
import akka.actor.typed.scaladsl.Behaviors


object HelloActor {
  final case class SayHello(
      name: String,
      replyTo: ActorRef[String],
  )


  def create(): Behavior[HelloActor.SayHello] = {
    Behaviors.setup(context => new HelloActor(context))
  }
}


final class HelloActor private (
    context: ActorContext[HelloActor.SayHello],
) extends AbstractBehavior(context) {
  import HelloActor._


  def onMessage(msg: SayHello) = {
    msg.replyTo ! s"Hello, ${msg.name}"
    this
  }
}

📘 Java FP

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Behaviors;


public final class HelloActor {


  public static final class SayHello {
    public final String name;
    public final ActorRef<String> replyTo;


    public SayHello(String name, ActorRef<String> replyTo) {
      this.name = name;
      this.replyTo = replyTo;
    }
  }


  public static Behavior<HelloActor.SayHello> create() {
    return Behaviors.receiveMessage(
        (SayHello message) -> {
          message.replyTo.tell("Hello, " + message.name);
          return Behaviors.same();
        });
  }
}

You can also try this code with Online Java Compiler

Run Code

📘 Java OO

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;


public final class HelloActor extends AbstractBehavior<HelloActor.SayHello> {


  public static final class SayHello {
    public final String name;
    public final ActorRef<String> replyTo;


    public SayHello(String name, ActorRef<String> replyTo) {
      this.name = name;
      this.replyTo = replyTo;
    }
  }


  public static Behavior<HelloActor.SayHello> create() {
    return Behaviors.setup((ctx) -> new HelloActor(ctx));
  }


  private HelloActor(ActorContext<HelloActor.SayHello> context) {
    super(context);
  }


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


  private Behavior<SayHello> onHello(SayHello message) {
    message.replyTo.tell("Hello, " + message.name);
    return this;
  }
}

You can also try this code with Online Java Compiler

Run Code

Below is an example of an actor that depends on Play’s Configuration for returning configuration values:

📘 Scala FP

import akka.actor.typed.ActorRef
import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl.Behaviors
import com.google.inject.Provides
import play.api.Configuration
import play.api.libs.concurrent.ActorModule


object ConfiguredActor extends ActorModule {
  type Message = GetConfig


  final case class GetConfig(replyTo: ActorRef[String])


  @Provides
  def create(configuration: Configuration): Behavior[GetConfig] = {
    Behaviors.setup { _ =>
      val config = configuration.get[String]("my.config")
      Behaviors.receiveMessage[GetConfig] {
        case GetConfig(replyTo) =>
          replyTo ! config
          Behaviors.same
      }
    }
  }
}

📘 Scala OO

import akka.actor.typed.ActorRef
import akka.actor.typed.Behavior
import akka.actor.typed.scaladsl.AbstractBehavior
import akka.actor.typed.scaladsl.ActorContext
import akka.actor.typed.scaladsl.Behaviors
import javax.inject.Inject
import play.api.Configuration


object ConfiguredActor {
  final case class GetConfig(replyTo: ActorRef[String])


  def create(
      configuration: Configuration,
  ): Behavior[ConfiguredActor.GetConfig] = {
    Behaviors.setup { context =>
      new ConfiguredActor(context, configuration)
    }
  }
}


final class ConfiguredActor private (
    context: ActorContext[ConfiguredActor.GetConfig],
    configuration: Configuration,
) extends AbstractBehavior(context) {
  import ConfiguredActor._


  val config = configuration.get[String]("my.config")
  def onMessage(msg: GetConfig) = {
    msg.replyTo ! config
    this
  }
}

📘 Java FP

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import com.typesafe.config.Config;


import javax.inject.Inject;


public final class ConfiguredActor {


  public static final class GetConfig {
    public final ActorRef<String> replyTo;


    public GetConfig(ActorRef<String> replyTo) {
      this.replyTo = replyTo;
    }
  }


  public static Behavior<ConfiguredActor.GetConfig> create(Config config) {
    String myConfig = config.getString("my.config");
    return Behaviors.receiveMessage(
        (GetConfig message) -> {
          message.replyTo.tell(myConfig);
          return Behaviors.same();
        });
  }
}

You can also try this code with Online Java Compiler

Run Code

📘 Java OO

import akka.actor.typed.ActorRef;
import akka.actor.typed.Behavior;
import akka.actor.typed.javadsl.AbstractBehavior;
import akka.actor.typed.javadsl.ActorContext;
import akka.actor.typed.javadsl.Behaviors;
import akka.actor.typed.javadsl.Receive;
import com.typesafe.config.Config;


public final class ConfiguredActor extends AbstractBehavior<ConfiguredActor.GetConfig> {


  public static final class GetConfig {
    public final ActorRef<String> replyTo;


    public GetConfig(ActorRef<String> replyTo) {
      this.replyTo = replyTo;
    }
  }


  private final String config;


  public static Behavior<ConfiguredActor.GetConfig> create(Config config) {
    return Behaviors.setup((ctx) -> new ConfiguredActor(ctx, config));
  }


  private ConfiguredActor(ActorContext<ConfiguredActor.GetConfig> context, Config config) {
    super(context);
    this.config = config.getString("my.config");
  }


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


  private Behavior<GetConfig> onGetConfig(GetConfig message) {
    message.replyTo.tell(config);
    return this;
  }
}

You can also try this code with Online Java Compiler

Run Code

Dependency Injection

If our actor’s behavior has a variable state and is sometimes common in the object-oriented style, we need to ensure that we don’t share the same Behaviour instance for multiple Actorref s. Some general ways to avoid this problem are as follows:

📌 Consider a design without variable state

📌 Do not deluge the Behaviour instance by only exposing the Actorref instance, for example, by only binding the ActorRef

📌 If the objective is only to have a single instance of the actor, then make sure that both the Behaviour and ActorRef are singletons.

📌 Rather, there are meant to be multiple instances of the same actor, then make sure both ActorRef and Behaviour are named singletons in Guice by using @Named or .annotatedWith(Names.named(...)).

Compile-time dependency injection

Using compile time dependency injection for Akka Actor Typed needs creating the actor Behaviour value and using it to spawn the actor:

📘 Scala

import akka.actor.typed.scaladsl.adapter._
import play.api._
import play.api.routing.Router


final class AppComponents(context: ApplicationLoader.Context)
    extends BuiltInComponentsFromContext(context)
    with NoHttpFiltersComponents {
  val router = Router.empty


  val helloActor = {
    actorSystem.spawn(HelloActor.create(), "hello-actor")
  }
  val configuredActor = {
    val behavior = ConfiguredActor.create(configuration)
    actorSystem.spawn(behavior, "configured-actor")
  }


  val main = new Main(helloActor, configuredActor)
}

📘 Java

import akka.actor.typed.ActorRef;
import akka.actor.typed.javadsl.Adapter;
import play.ApplicationLoader;
import play.BuiltInComponentsFromContext;
import play.mvc.EssentialFilter;
import play.routing.Router;


import java.util.Collections;
import java.util.List;


public final class AppComponents extends BuiltInComponentsFromContext {


  public final ActorRef<HelloActor.SayHello> helloActor;
  public final ActorRef<ConfiguredActor.GetConfig> configuredActor;
  public final Main main;


  public AppComponents(ApplicationLoader.Context context) {
    super(context);
    helloActor = Adapter.spawn(actorSystem(), HelloActor.create(), "hello-actor");
    configuredActor =
        Adapter.spawn(actorSystem(), ConfiguredActor.create(config()), "configured-actor");
    main = new Main(helloActor, configuredActor);
  }


  @Override
  public Router router() {
    return Router.empty();
  }


  @Override
  public List<EssentialFilter> httpFilters() {
    return Collections.emptyList();
  }
}

You can also try this code with Online Java Compiler

Run Code

Runtime dependency injection

For runtime dependency injection, we need to use the “typed” method in AkkaGuiceSupport if we are using a functional programming style. For the object-oriented style, we must write a Provider for our ActorRef and bind it.

Following code shows injecting for a given component in your application :

📘Scala

import javax.inject.Inject
import javax.inject.Singleton
import akka.actor.typed.ActorRef


@Singleton final class Main @Inject() (
    val helloActor: ActorRef[HelloActor.SayHello],
    val configuredActor: ActorRef[ConfiguredActor.GetConfig],
)

📘Java

import akka.actor.typed.ActorRef;

import javax.inject.Inject;
import javax.inject.Singleton;

@Singleton
public final class Main {
  public final ActorRef<HelloActor.SayHello> helloActor;
  public final ActorRef<ConfiguredActor.GetConfig> configuredActor;

 @Inject
  public Main(
      ActorRef<HelloActor.SayHello> helloActor,
      ActorRef<ConfiguredActor.GetConfig> configuredActor) {
    this.helloActor = helloActor;
    this.configuredActor = configuredActor;
  }
}

You can also try this code with Online Java Compiler

Run Code

We can also define a Guice Module like the below:

📘Scala FP

import com.google.inject.AbstractModule
import play.api.libs.concurrent.AkkaGuiceSupport

object AppModule extends AbstractModule with AkkaGuiceSupport {
  override def configure() = {
    bindTypedActor(HelloActor.create(), "hello-actor")  // uses "create" method
    bindTypedActor(ConfiguredActor, "configured-actor") // uses the object itself
  }
}

📘Scala OO

import akka.actor.ActorSystem
import akka.actor.typed.scaladsl.adapter._
import akka.actor.typed.ActorRef
import com.google.inject.AbstractModule
import com.google.inject.Provider
import com.google.inject.TypeLiteral
import javax.inject.Inject
import play.api.Configuration
import play.api.libs.concurrent.AkkaGuiceSupport


object AppModule extends AbstractModule with AkkaGuiceSupport {
  override def configure() = {
    bindTypedActor(HelloActor.create(), "hello-actor")
    bind(new TypeLiteral[ActorRef[ConfiguredActor.GetConfig]]() {})
      .toProvider(classOf[ConfiguredActorProvider])
      .asEagerSingleton()
  }


  private class ConfiguredActorProvider @Inject() (
      actorSystem: ActorSystem,
      configuration: Configuration,
  ) extends Provider[ActorRef[ConfiguredActor.GetConfig]] {
    def get() = {
      val behavior = ConfiguredActor.create(configuration)
      actorSystem.spawn(behavior, "configured-actor")
    }
  }
}

📘Java FP

import akka.actor.ActorSystem;
import akka.actor.typed.javadsl.Adapter;
import akka.actor.typed.ActorRef;
import com.google.inject.AbstractModule;
import com.google.inject.Provider;
import com.google.inject.TypeLiteral;
import com.typesafe.config.Config;
import javax.inject.Inject;

public class AppModule extends AbstractModule {


  @Override
  protected void configure() {
    bind(new TypeLiteral<ActorRef<HelloActor.SayHello>>() {})
        .toProvider(HelloActorProvider.class)
        .asEagerSingleton();
    bind(new TypeLiteral<ActorRef<ConfiguredActor.GetConfig>>() {})
        .toProvider(ConfiguredActorProvider.class)
        .asEagerSingleton();
  }
  public static class HelloActorProvider implements Provider<ActorRef<HelloActor.SayHello>> {
    private final ActorSystem actorSystem;


    @Inject
    public HelloActorProvider(ActorSystem actorSystem) {
      this.actorSystem = actorSystem;
    }


    @Override
    public ActorRef<HelloActor.SayHello> get() {
      return Adapter.spawn(actorSystem, HelloActor.create(), "hello-actor");
    }
  }


  public static class ConfiguredActorProvider
      implements Provider<ActorRef<ConfiguredActor.GetConfig>> {


    private final ActorSystem actorSystem;
    private final Config config;


    @Inject
    public ConfiguredActorProvider(ActorSystem actorSystem, Config config) {
      this.actorSystem = actorSystem;
      this.config = config;
    }


    @Override
    public ActorRef<ConfiguredActor.GetConfig> get() {
      return Adapter.spawn(actorSystem, ConfiguredActor.create(config), "configured-actor");
    }
  }
}

You can also try this code with Online Java Compiler

Run Code

📘Java OO

import akka.actor.ActorSystem;
import akka.actor.typed.javadsl.Adapter;
import akka.actor.typed.ActorRef;
import com.google.inject.AbstractModule;
import com.google.inject.Provider;
import com.google.inject.TypeLiteral;
import com.typesafe.config.Config;
import javax.inject.Inject;

public class AppModule extends AbstractModule {
  @Override
  protected void configure() {
    bind(new TypeLiteral<ActorRef<HelloActor.SayHello>>() {})
        .toProvider(HelloActorProvider.class)
        .asEagerSingleton();
    bind(new TypeLiteral<ActorRef<ConfiguredActor.GetConfig>>() {})
        .toProvider(ConfiguredActorProvider.class)
        .asEagerSingleton();
  }

  public static class HelloActorProvider implements Provider<ActorRef<HelloActor.SayHello>> {
    private final ActorSystem actorSystem;

    @Inject
    public HelloActorProvider(ActorSystem actorSystem) {
      this.actorSystem = actorSystem;
    }
    @Override
    public ActorRef<HelloActor.SayHello> get() {
      return Adapter.spawn(actorSystem, HelloActor.create(), "hello-actor");
    }
  }
  
   public static class ConfiguredActorProvider
      implements Provider<ActorRef<ConfiguredActor.GetConfig>> {

    private final ActorSystem actorSystem;
    private final Config config;

     @Inject
    public ConfiguredActorProvider(ActorSystem actorSystem, Config config) {
      this.actorSystem = actorSystem;
      this.config = config;
    }

    @Override
    public ActorRef<ConfiguredActor.GetConfig> get() {
      return Adapter.spawn(actorSystem, ConfiguredActor.create(config), "configured-actor");
    }
  }
}

You can also try this code with Online Java Compiler

Run Code

Cluster Sharding for Akka Typed (incubating)

Play provides an incubating module for integration with Akka Cluster Sharding Typed, and to enable this module, we need to add the following dependencies to our build:

Usage

After properly including Cluster Sharding as a dependency, we can obtain an instance by using dependency injection. We have provided some helpers for both compile-time and runtime dependency injection. Here, notice that Play is on;y providing the DI mechanism.

The class instance that will be made available for injection is as follows:

Runtime dependency injection

Runtime dependency injection works as any other runtime DI module in Play, enabling that module automatically by adding the dependency, and an instance is available for injection.

Compile-time dependency injection

If we are using compile-time DI, we can get have access to the ClusterSharding by using the components as below :

📘Java

import play.Application;
import play.ApplicationLoader;
import play.BuiltInComponentsFromContext;
import play.controllers.AssetsComponents;
import play.routing.Router;
import play.cluster.sharding.typed.ClusterShardingComponents;
import play.filters.components.HttpFiltersComponents;


public class ComponentsWithClusterSharding extends BuiltInComponentsFromContext
    implements ClusterShardingComponents, AssetsComponents, HttpFiltersComponents {


  public ComponentsWithClusterSharding(ApplicationLoader.Context context) {
    super(context);
  }


  @Override
  public Router router() {
    return Router.empty();
  }
}

You can also try this code with Online Java Compiler

Run Code

📘Scala

import play.api._
import play.api.ApplicationLoader.Context
import play.api.routing.Router
import play.api.cluster.sharding.typed.ClusterShardingComponents


class MyApplicationLoader extends ApplicationLoader {
  def load(context: Context) = {
    new ComponentsWithClusterSharding(context).application
  }
}


class ComponentsWithClusterSharding(context: Context)
    extends BuiltInComponentsFromContext(context)
    with play.filters.HttpFiltersComponents
    with ClusterShardingComponents {
  lazy val router = Router.empty
}

Frequently Asked Questions

Describe the Akka cluster.

A fault-tolerant decentralized peer-to-peer Cluster Membership Service without a single point of failure or bottleneck is offered by Akka Cluster.

Why is the Akka framework used?

Akka framework helps to build software systems with high performance, and reliability, without sacrificing the developer’s productivity and joy.

What is a JVM application?

The JVM provides a portable execution environment for Java-based applications while managing system memory.

What are the actors in Akka?

Actors are objects that receive messages and respond to them by taking appropriate action.

Which statement terminates the JVM?

The JVM is shut down (by sending SIGTERM) Runtime. exit() or System. exit() is called by one of the threads.

Conclusion

In this article, we discussed Integrating with Akka Typed & cluster Sharding in detail. We started with introducing the Akka framework, and then we saw Integrating with Akka Typed with the help of both Java and Scala. We saw dependency injections and then cluster sharding for Akka Typed.

Visit our Guided Path in Coding Ninjas Studio to learn about its related topics. If you are preparing for an interview, visit our Interview Experience Section and interview bundle for placement preparations. Upskill yourself in Python, Kivy, Backend Web Technologies, SQL, MongoDB, Data Structures and Algorithms, JavaScript, System Design, and much more!