Implementing Event Sourced Entities

@Override
public Effect<Empty> addItem(
    ShoppingCartDomain.Cart currentState,
    ShoppingCartApi.AddLineItem command) {
  if (currentState.getCheckedOut())
    return effects().error("Cart is already checked out.");
  if (command.getQuantity() <= 0) { (1)
    return effects().error("Quantity for item " + command.getProductId() + " must be greater than zero.");
  }

  ShoppingCartDomain.ItemAdded event = (2)
      ShoppingCartDomain.ItemAdded.newBuilder()
          .setItem(
              ShoppingCartDomain.LineItem.newBuilder()
                  .setProductId(command.getProductId())
                  .setName(command.getName())
                  .setQuantity(command.getQuantity())
                  .build())
          .build();

  return effects()
          .emitEvent(event) (3)
          .thenReply(newState -> Empty.getDefaultInstance()); (4)
}

override def addItem(currentState: Cart, addLineItem: shoppingcart.AddLineItem): EventSourcedEntity.Effect[Empty] =
  if (currentState.checkedOut)
    effects.error("Cart is already checked out")
  else if (addLineItem.quantity <= 0)
    effects.error(s"Quantity for item ${addLineItem.productId} must be greater than zero.") (1)
  else {
    val event = ItemAdded( (2)
      Some(LineItem(productId = addLineItem.productId, name = addLineItem.name, quantity = addLineItem.quantity)))
    effects
      .emitEvent(event) (3)
      .thenReply(_ => Empty.defaultInstance) (4)
  }

@Override
public ShoppingCartDomain.Cart itemAdded(
    ShoppingCartDomain.Cart currentState,
    ShoppingCartDomain.ItemAdded itemAdded) {
  ShoppingCartDomain.LineItem item = itemAdded.getItem();
  ShoppingCartDomain.LineItem lineItem = updateItem(item, currentState);
  List<ShoppingCartDomain.LineItem> lineItems =
      removeItemByProductId(currentState, item.getProductId());
  lineItems.add(lineItem);
  lineItems.sort(Comparator.comparing(ShoppingCartDomain.LineItem::getProductId));
  return ShoppingCartDomain.Cart.newBuilder().addAllItems(lineItems).build();
}

private ShoppingCartDomain.LineItem updateItem(
    ShoppingCartDomain.LineItem item, ShoppingCartDomain.Cart cart) {
  return findItemByProductId(cart, item.getProductId())
      .map(li -> li.toBuilder().setQuantity(li.getQuantity() + item.getQuantity()).build())
      .orElse(item);
}

private Optional<ShoppingCartDomain.LineItem> findItemByProductId(
    ShoppingCartDomain.Cart cart, String productId) {
  Predicate<ShoppingCartDomain.LineItem> lineItemExists =
      lineItem -> lineItem.getProductId().equals(productId);
  return cart.getItemsList().stream().filter(lineItemExists).findFirst();
}

private List<ShoppingCartDomain.LineItem> removeItemByProductId(
    ShoppingCartDomain.Cart cart, String productId) {
  return cart.getItemsList().stream()
      .filter(lineItem -> !lineItem.getProductId().equals(productId))
      .collect(Collectors.toList());
}

override def itemAdded(currentState: Cart, itemAdded: ItemAdded): Cart = {
  val cart = currentState.items.map(lineItem => lineItem.productId -> lineItem).toMap
  val item = cart.get(itemAdded.getItem.productId) match {
    case Some(existing) => existing.copy(quantity = existing.quantity + itemAdded.getItem.quantity)
    case None           => itemAdded.getItem
  }
  val updatedCart = cart + (item.productId -> item)
  currentState.withItems(updatedCart.values.toSeq)
}

@Override
public Effect<ShoppingCartApi.Cart> getCart(
    ShoppingCartDomain.Cart currentState, (1)
    ShoppingCartApi.GetShoppingCart command) {
  List<ShoppingCartApi.LineItem> apiItems =
      currentState.getItemsList().stream()
          .map(this::convert)
          .sorted(Comparator.comparing(ShoppingCartApi.LineItem::getProductId))
          .collect(Collectors.toList());
  ShoppingCartApi.Cart apiCart =
          ShoppingCartApi.Cart.newBuilder().addAllItems(apiItems)
              .setCheckedOut(currentState.getCheckedOut())
              .build(); (2)
  return effects().reply(apiCart);
}

private ShoppingCartApi.LineItem convert(ShoppingCartDomain.LineItem item) {
  return ShoppingCartApi.LineItem.newBuilder()
          .setProductId(item.getProductId())
          .setName(item.getName())
          .setQuantity(item.getQuantity())
          .build();
}

override def getCart(
    currentState: Cart, (1)
    getShoppingCart: shoppingcart.GetShoppingCart): EventSourcedEntity.Effect[shoppingcart.Cart] = {
  val apiItems = currentState.items.map(convertToApi).sortBy(_.productId)
  val apiCart = shoppingcart.Cart(apiItems, currentState.checkedOut) (2)
  effects.reply(apiCart)
}

private def convertToApi(item: LineItem): shoppingcart.LineItem =
  shoppingcart.LineItem(productId = item.productId, name = item.name, quantity = item.quantity)

package com.example.shoppingcart.domain;

import kalix.javasdk.testkit.EventSourcedResult;
import com.example.shoppingcart.ShoppingCartApi;
import com.google.protobuf.Empty;
import org.junit.jupiter.api.Test;

import java.util.NoSuchElementException;

import static org.junit.jupiter.api.Assertions.assertEquals;
import static org.junit.jupiter.api.Assertions.assertThrows;

public class ShoppingCartTest {

    @Test
    public void addItemTest() {

        ShoppingCartTestKit testKit = ShoppingCartTestKit.of(ShoppingCart::new); (1)

        ShoppingCartApi.AddLineItem apples = ShoppingCartApi.AddLineItem.newBuilder().setProductId("idA")
                .setName("apples").setQuantity(1).build();
        EventSourcedResult<Empty> addingApplesResult = testKit.addItem(apples); (2)

        ShoppingCartApi.AddLineItem bananas = ShoppingCartApi.AddLineItem.newBuilder().setProductId("idB")
                .setName("bananas").setQuantity(2).build();
        testKit.addItem(bananas); (3)

        assertEquals(1, addingApplesResult.getAllEvents().size()); (4)
        assertEquals(2, testKit.getAllEvents().size()); (5)

        ShoppingCartDomain.ItemAdded addedApples = addingApplesResult.getNextEventOfType(ShoppingCartDomain.ItemAdded.class); (6)
        assertEquals("apples", addedApples.getItem().getName());
        assertThrows(NoSuchElementException.class, () ->  addingApplesResult.getNextEventOfType(ShoppingCartDomain.ItemAdded.class)); (7)
        assertEquals(Empty.getDefaultInstance(), addingApplesResult.getReply()); (8)

        ShoppingCartDomain.LineItem expectedApples = ShoppingCartDomain.LineItem.newBuilder().setProductId("idA")
                .setName("apples").setQuantity(1).build();        
        ShoppingCartDomain.LineItem expectedBananas = ShoppingCartDomain.LineItem.newBuilder().setProductId("idB")
                .setName("bananas").setQuantity(2).build();
        ShoppingCartDomain.Cart expectedState = ShoppingCartDomain.Cart.newBuilder()
                .addItems(expectedApples)
                .addItems(expectedBananas)
                .build();
        assertEquals(expectedState, testKit.getState()); (9)
    }
}

package com.example.shoppingcart.domain

import kalix.scalasdk.eventsourcedentity.EventSourcedEntity
import kalix.scalasdk.testkit.EventSourcedResult
import com.example.shoppingcart
import com.example.shoppingcart.AddLineItem
import com.google.protobuf.empty.Empty
import org.scalatest.matchers.should.Matchers
import org.scalatest.wordspec.AnyWordSpec

class ShoppingCartSpec extends AnyWordSpec with Matchers {
  "The ShoppingCart" should {

    "correctly process commands of type AddItem" in {
      val testKit = ShoppingCartTestKit(new ShoppingCart(_)) (1)
      val apples = AddLineItem(productId = "idA", name = "apples", quantity = 1)
      val addingApplesResult = testKit.addItem(apples) (2)

      val bananas = AddLineItem(productId = "idB", name = "bananas", quantity = 2)
      testKit.addItem(bananas) (3)

      addingApplesResult.events should have size 1 (4)
      testKit.allEvents should have size 2 (5)

      val addedApples = addingApplesResult.nextEvent[ItemAdded] (6)
      addedApples.getItem.name shouldBe "apples"
      intercept[NoSuchElementException] { (7)
        addingApplesResult.nextEvent[ItemAdded]
      }
      addingApplesResult.reply shouldBe Empty.defaultInstance (8)

      val expectedState = Cart(Seq(
        LineItem(productId = "idA", name = "apples", quantity = 1),
        LineItem(productId = "idB", name = "bananas", quantity = 2)
      ))
      testKit.currentState shouldBe expectedState (9)
    }
  }
}

import kalix.javasdk.testkit.junit.jupiter.KalixTestKitExtension;
// ...

public class ShoppingCartIntegrationTest {

  /**
   * The test kit starts both the service container and the Kalix Runtime.
   */
  @RegisterExtension
  public static final KalixTestKitExtension testKit =
    new KalixTestKitExtension(Main.createKalix()); (1)

  private final ShoppingCartService client;

  public ShoppingCartIntegrationTest() {
    this.client = testKit.getGrpcClient(ShoppingCartService.class); (2)
  }

  @Test
  public void addItemsToCart() throws Exception { (3)
    addItem("cart2", "a", "Apple", 1);
    addItem("cart2", "b", "Banana", 2);
    addItem("cart2", "c", "Cantaloupe", 3);
    ShoppingCartApi.Cart cart = getCart("cart2");
    assertEquals(3, cart.getItemsCount(), "shopping cart should have 3 items");
    assertEquals(
      cart.getItemsList(),
      List.of(item("a", "Apple", 1), item("b", "Banana", 2), item("c", "Cantaloupe", 3)),
      "shopping cart should have expected items"
    );
  }

  ShoppingCartApi.Cart getCart(String cartId) throws Exception {
    return client
      .getCart(ShoppingCartApi.GetShoppingCart.newBuilder().setCartId(cartId).build())
      .toCompletableFuture()
      .get();
  }

  void addItem(String cartId, String productId, String name, int quantity) throws Exception {
    client
      .addItem(
        ShoppingCartApi.AddLineItem.newBuilder()
          .setCartId(cartId)
          .setProductId(productId)
          .setName(name)
          .setQuantity(quantity)
          .build())
      .toCompletableFuture()
      .get();
  }

  ShoppingCartApi.LineItem item(String productId, String name, int quantity) {
    return ShoppingCartApi.LineItem.newBuilder()
      .setProductId(productId)
      .setName(name)
      .setQuantity(quantity)
      .build();
  }
}

import kalix.scalasdk.testkit.KalixTestKit
// ...

class ShoppingCartServiceIntegrationSpec
    extends AnyWordSpec
    with Matchers
    with BeforeAndAfterAll
    with ScalaFutures {

  implicit val patience: PatienceConfig =
    PatienceConfig(Span(5, Seconds), Span(500, Millis))

  private val testKit = KalixTestKit(Main.createKalix()).start() (1)
  import testKit.executionContext

  "ShoppingCartService" must {
    val client = testKit.getGrpcClient(classOf[ShoppingCartService]) (2)

    "add items to shopping cart" in {
      val cartId = "cart1"

      val updatedCart = for {
        _ <- client.addItem(AddLineItem(cartId, "shirt", "Shirt", 1))
        done <- client.addItem(AddLineItem(cartId, "sweat", "Sweat Shirt", 2))
      } yield done

      updatedCart.futureValue

      client.getCart(GetShoppingCart(cartId)).futureValue.items shouldBe (3)
        Seq(LineItem("shirt", "Shirt", 1), LineItem("sweat", "Sweat Shirt", 2))
    }

  }

  override def afterAll() = { (4)
    testKit.stop()
    super.afterAll()
  }
}