Setup your application

Add the Akka Projections core library to a new project. This isn’t strictly required, because as we add other dependencies in the following steps it will transitively include core as a dependency, but it never hurts to be explicit.

sbt

libraryDependencies += "com.lightbend.akka" %% "akka-projection-core" % "1.1.0"

Maven

<properties>
  <scala.binary.version>2.13</scala.binary.version>
</properties>
<dependencies>
  <dependency>
    <groupId>com.lightbend.akka</groupId>
    <artifactId>akka-projection-core_${scala.binary.version}</artifactId>
    <version>1.1.0</version>
  </dependency>
</dependencies>

Gradle

def versions = [
  ScalaBinary: "2.13"
]
dependencies {
  implementation "com.lightbend.akka:akka-projection-core_${versions.ScalaBinary}:1.1.0"
}

Define the event type protocol that will represent each Envelope streamed from the Source Provider. Add ShoppingCartEvents to your project:

Scala

package docs.guide

import java.time.Instant

object ShoppingCartEvents {

  /**
   * This interface defines all the events that the ShoppingCart supports.
   */
  sealed trait Event extends CborSerializable {
    def cartId: String
  }

  sealed trait ItemEvent extends Event {
    def itemId: String
  }

  final case class ItemAdded(cartId: String, itemId: String, quantity: Int) extends ItemEvent
  final case class ItemRemoved(cartId: String, itemId: String, oldQuantity: Int) extends ItemEvent
  final case class ItemQuantityAdjusted(cartId: String, itemId: String, newQuantity: Int, oldQuantity: Int)
      extends ItemEvent
  final case class CheckedOut(cartId: String, eventTime: Instant) extends Event
}

Java

package jdocs.guide;

import java.time.Instant;

public class ShoppingCartEvents {
  public interface Event extends CborSerializable {
    String getCartId();
  }

  public interface ItemEvent extends Event {
    String getItemId();
  }

  public static final class ItemAdded implements ItemEvent {
    public final String cartId;
    public final String itemId;
    public final int quantity;

    public ItemAdded(String cartId, String itemId, int quantity) {
      this.cartId = cartId;
      this.itemId = itemId;
      this.quantity = quantity;
    }

    public String getCartId() {
      return this.cartId;
    }

    public String getItemId() {
      return this.itemId;
    }
  }

  public static final class ItemRemoved implements ItemEvent {
    public final String cartId;
    public final String itemId;
    public final int oldQuantity;

    public ItemRemoved(String cartId, String itemId, int oldQuantity) {
      this.cartId = cartId;
      this.itemId = itemId;
      this.oldQuantity = oldQuantity;
    }

    public String getCartId() {
      return this.cartId;
    }

    public String getItemId() {
      return this.itemId;
    }
  }

  public static final class ItemQuantityAdjusted implements ItemEvent {
    public final String cartId;
    public final String itemId;
    public final int newQuantity;
    public final int oldQuantity;

    public ItemQuantityAdjusted(String cartId, String itemId, int newQuantity, int oldQuantity) {
      this.cartId = cartId;
      this.itemId = itemId;
      this.newQuantity = newQuantity;
      this.oldQuantity = oldQuantity;
    }

    public String getCartId() {
      return this.cartId;
    }

    public String getItemId() {
      return this.itemId;
    }
  }

  public static final class CheckedOut implements Event {
    public final String cartId;
    public final Instant eventTime;

    public CheckedOut(String cartId, Instant eventTime) {
      this.cartId = cartId;
      this.eventTime = eventTime;
    }

    public String getCartId() {
      return this.cartId;
    }
  }
}

To enable serialization and deserialization of events with Akka Persistence it’s necessary to define a base type for your event type hierarchy. In this guide we are using Jackson Serialization. Add the CborSerializable base type to your project:

Scala

package docs.guide

/**
 * Marker trait for serialization with Jackson CBOR
 */
trait CborSerializable

Java

package jdocs.guide;

/** Marker trait for serialization with Jackson CBOR */
public interface CborSerializable {}

Configure the CborSerializable type to use jackson-cbor configuration in your application.conf. We will add this configuration when Akka Persistence configuration is setup in the Choosing a SourceProvider section of the guide.

Scala

akka.actor.serialization-bindings {
  "docs.guide.CborSerializable" = jackson-cbor
}

Java

akka.actor.serialization-bindings {
  "jdocs.guide.CborSerializable" = jackson-cbor
}

Define the persistence tags to be used in your project. Note that partitioned tags will be used later when running the projection in Akka Cluster. Add ShoppingCartTags to your project:

Scala

package docs.guide

object ShoppingCartTags {
  val Single = "shopping-cart"
  val Tags = Vector("carts-0", "carts-1", "carts-2")
}

Java

package jdocs.guide;

public class ShoppingCartTags {
  public static String SINGLE = "shopping-cart";
  public static String[] TAGS = {"carts-0", "carts-1", "carts-2"};
}

Create the ShoppingCartApp with an akka.actor.typed.ActorSystem (API: ActorSystemActorSystem) for Projections to use. Create an empty Guardian Actor (the root Actor of the ActorSystem). We will populate this Actor in the following steps of the guide. Note that we are using the docs.scaladsljdocs.scaladsl package. You may use any package, but we include this package in snippets throughout the guide.

Scala

package docs.guide

import akka.actor.typed.ActorSystem
import akka.actor.typed.scaladsl.Behaviors
import akka.projection.ProjectionBehavior
import akka.projection.eventsourced.EventEnvelope
import com.typesafe.config.ConfigFactory

object ShoppingCartApp extends App {
  val config = ConfigFactory.load("guide-shopping-cart-app.conf")

  ActorSystem(
    Behaviors.setup[String] { context =>
      val system = context.system

      // ...

      Behaviors.empty
    },
    "ShoppingCartApp",
    config)
}

Java

package jdocs.guide;

import akka.actor.typed.ActorSystem;
import akka.actor.typed.javadsl.Behaviors;
import akka.projection.ProjectionBehavior;
import akka.projection.eventsourced.EventEnvelope;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;

public class ShoppingCartApp {
  public static void main(String[] args) throws Exception {
    Config config = ConfigFactory.load("guide-shopping-cart-app.conf");

    ActorSystem.create(
        Behaviors.setup(
            context -> {
              ActorSystem<Void> system = context.getSystem();

              // ...

              return Behaviors.empty();
            }),
        "ShoppingCartApp",
        config);
  }
}