Use Cassandra instead of PostgreSQL

package shopping.cart;

import akka.Done;
import akka.projection.eventsourced.EventEnvelope;
import akka.projection.javadsl.Handler;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.CompletionStage;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public final class ItemPopularityProjectionHandler
    extends Handler<EventEnvelope<ShoppingCart.Event>> { (1)
  private final Logger logger = LoggerFactory.getLogger(getClass());
  private final String tag;
  private final ItemPopularityRepository repo;

  public ItemPopularityProjectionHandler(String tag, ItemPopularityRepository repo) {
    this.tag = tag;
    this.repo = repo;
  }

  @Override
  public CompletionStage<Done> process(EventEnvelope<ShoppingCart.Event> envelope) (2)
      throws Exception {
    ShoppingCart.Event event = envelope.event();

    if (event instanceof ShoppingCart.ItemAdded) { (3)
      ShoppingCart.ItemAdded added = (ShoppingCart.ItemAdded) event;
      CompletionStage<Done> result = this.repo.update(added.itemId, added.quantity);
      result.thenAccept(done -> logItemCount(added.itemId));
      return result;
    } else {
      // skip all other events, such as `CheckedOut`
      return CompletableFuture.completedFuture(Done.getInstance());
    }
  }

  private void logItemCount(String itemId) {
    repo.getItem(itemId)
        .thenAccept(
            optCount ->
                logger.info(
                    "ItemPopularityProjectionHandler({}) item popularity for '{}': [{}]",
                    this.tag,
                    itemId,
                    optCount.orElse(0L)));
  }
}

package shopping.cart

import scala.concurrent.ExecutionContext
import scala.concurrent.Future

import akka.Done
import akka.actor.typed.ActorSystem
import akka.projection.eventsourced.EventEnvelope
import akka.projection.scaladsl.Handler
import org.slf4j.LoggerFactory

class ItemPopularityProjectionHandler(
    tag: String,
    system: ActorSystem[_],
    repo: ItemPopularityRepository)
    extends Handler[EventEnvelope[ShoppingCart.Event]]() { (1)

  private val log = LoggerFactory.getLogger(getClass)
  private implicit val ec: ExecutionContext =
    system.executionContext

  override def process(
      envelope: EventEnvelope[ShoppingCart.Event]): Future[Done] = { (2)
    envelope.event match { (3)
      case ShoppingCart.ItemAdded(_, itemId, quantity) =>
        val result = repo.update(itemId, quantity)
        result.foreach(_ => logItemCount(itemId))
        result


      case _: ShoppingCart.CheckedOut =>
        Future.successful(Done)
    }
  }

  private def logItemCount(itemId: String): Unit = {
    repo.getItem(itemId).foreach { optCount =>
      log.info(
        "ItemPopularityProjectionHandler({}) item popularity for '{}': [{}]",
        tag,
        itemId,
        optCount.getOrElse(0))
    }
  }

}

package shopping.cart;

import akka.Done;
import java.util.Optional;
import java.util.concurrent.CompletionStage;

public interface ItemPopularityRepository {
  CompletionStage<Done> update(String itemId, int delta);

  CompletionStage<Optional<Long>> getItem(String itemId);
}

package shopping.cart

import scala.concurrent.Future
import akka.Done

trait ItemPopularityRepository {
  def update(itemId: String, delta: Int): Future[Done]
  def getItem(itemId: String): Future[Option[Long]]
}

package shopping.cart;

import akka.Done;
import akka.stream.alpakka.cassandra.javadsl.CassandraSession;
import java.util.Optional;
import java.util.concurrent.CompletionStage;

public final class ItemPopularityRepositoryImpl implements ItemPopularityRepository {

  static final String POPULARITY_TABLE = "item_popularity";

  private final CassandraSession session;
  private final String table;

  public ItemPopularityRepositoryImpl(CassandraSession session, String keyspace) {
    this.session = session;
    this.table = keyspace + "." + POPULARITY_TABLE;
  }

  @Override
  public CompletionStage<Done> update(String itemId, int delta) {
    return session.executeWrite(
        "UPDATE " + table + " SET count = count + ? WHERE item_id = ?",
        Long.valueOf(delta),
        itemId);
  }

  @Override
  public CompletionStage<Optional<Long>> getItem(String itemId) {
    return session
        .selectOne("SELECT item_id, count FROM " + table + " WHERE item_id = ?", itemId)
        .thenApply(opt -> opt.map(row -> row.getLong("count")));
  }
}

package shopping.cart

import scala.concurrent.Future
import akka.Done
import scala.concurrent.ExecutionContext
import akka.stream.alpakka.cassandra.scaladsl.CassandraSession

object ItemPopularityRepositoryImpl {
  val popularityTable = "item_popularity"
}

class ItemPopularityRepositoryImpl(session: CassandraSession, keyspace: String)(
    implicit val ec: ExecutionContext)
    extends ItemPopularityRepository {
  import ItemPopularityRepositoryImpl.popularityTable

  override def update(itemId: String, delta: Int): Future[Done] = {
    session.executeWrite(
      s"UPDATE $keyspace.$popularityTable SET count = count + ? WHERE item_id = ?",
      java.lang.Long.valueOf(delta),
      itemId)
  }

  override def getItem(itemId: String): Future[Option[Long]] = {
    session
      .selectOne(
        s"SELECT item_id, count FROM $keyspace.$popularityTable WHERE item_id = ?",
        itemId)
      .map(opt => opt.map(row => row.getLong("count").longValue()))
  }
}

package shopping.cart;

import akka.actor.typed.ActorSystem;
import akka.cluster.sharding.typed.ShardedDaemonProcessSettings;
import akka.cluster.sharding.typed.javadsl.ShardedDaemonProcess;
import akka.persistence.cassandra.query.javadsl.CassandraReadJournal;
import akka.persistence.query.Offset;
import akka.projection.ProjectionBehavior;
import akka.projection.ProjectionId;
import akka.projection.cassandra.javadsl.CassandraProjection;
import akka.projection.eventsourced.EventEnvelope;
import akka.projection.eventsourced.javadsl.EventSourcedProvider;
import akka.projection.javadsl.AtLeastOnceProjection;
import akka.projection.javadsl.SourceProvider;
import java.util.Optional;

public final class ItemPopularityProjection {

  private ItemPopularityProjection() {}

  public static void init(ActorSystem<?> system, ItemPopularityRepository repository) {
    ShardedDaemonProcess.get(system)
        .init( (1)
            ProjectionBehavior.Command.class,
            "ItemPopularityProjection",
            ShoppingCart.TAGS.size(),
            index -> ProjectionBehavior.create(createProjectionFor(system, repository, index)),
            ShardedDaemonProcessSettings.create(system),
            Optional.of(ProjectionBehavior.stopMessage()));
  }

  private static AtLeastOnceProjection<Offset, EventEnvelope<ShoppingCart.Event>>
      createProjectionFor(ActorSystem<?> system, ItemPopularityRepository repository, int index) {
    String tag = ShoppingCart.TAGS.get(index); (2)

    SourceProvider<Offset, EventEnvelope<ShoppingCart.Event>> sourceProvider = (3)
        EventSourcedProvider.eventsByTag(
            system,
            CassandraReadJournal.Identifier(), (4)
            tag);

    return CassandraProjection.atLeastOnce( (5)
        ProjectionId.of("ItemPopularityProjection", tag),
        sourceProvider,
        () -> new ItemPopularityProjectionHandler(tag, repository)); (6)
  }
}

package shopping.cart

import akka.actor.typed.ActorSystem
import akka.cluster.sharding.typed.ShardedDaemonProcessSettings
import akka.cluster.sharding.typed.scaladsl.ShardedDaemonProcess
import akka.persistence.cassandra.query.scaladsl.CassandraReadJournal
import akka.persistence.query.Offset
import akka.projection.ProjectionBehavior
import akka.projection.ProjectionId
import akka.projection.cassandra.scaladsl.CassandraProjection
import akka.projection.eventsourced.EventEnvelope
import akka.projection.eventsourced.scaladsl.EventSourcedProvider
import akka.projection.scaladsl.AtLeastOnceProjection
import akka.projection.scaladsl.SourceProvider

object ItemPopularityProjection {
  def init(
      system: ActorSystem[_],
      repository: ItemPopularityRepository): Unit = {
    ShardedDaemonProcess(system).init( (1)
      name = "ItemPopularityProjection",
      ShoppingCart.tags.size,
      index =>
        ProjectionBehavior(createProjectionFor(system, repository, index)),
      ShardedDaemonProcessSettings(system),
      Some(ProjectionBehavior.Stop))
  }

  private def createProjectionFor(
      system: ActorSystem[_],
      repository: ItemPopularityRepository,
      index: Int)
      : AtLeastOnceProjection[Offset, EventEnvelope[ShoppingCart.Event]] = {
    val tag = ShoppingCart.tags(index) (2)

    val sourceProvider
        : SourceProvider[Offset, EventEnvelope[ShoppingCart.Event]] = (3)
      EventSourcedProvider.eventsByTag[ShoppingCart.Event](
        system = system,
        readJournalPluginId = CassandraReadJournal.Identifier, (4)
        tag = tag)

    CassandraProjection.atLeastOnce( (5)
      projectionId = ProjectionId("ItemPopularityProjection", tag),
      sourceProvider,
      handler = () =>
        new ItemPopularityProjectionHandler(tag, system, repository) (6)
    )
  }

}

CassandraSession session =
    CassandraSessionRegistry.get(system).sessionFor("akka.persistence.cassandra"); (1)
// use same keyspace for the item_popularity table as the offset store
Config config = system.settings().config();
String itemPopularityKeyspace =
    config.getString("akka.projection.cassandra.offset-store.keyspace");
ItemPopularityRepository itemPopularityRepository =
    new ItemPopularityRepositoryImpl(session, itemPopularityKeyspace); (2)

ItemPopularityProjection.init(system, itemPopularityRepository); (3)

import akka.stream.alpakka.cassandra.scaladsl.CassandraSessionRegistry

    val session = CassandraSessionRegistry(system).sessionFor(
      "akka.persistence.cassandra"
    ) (1)
    // use same keyspace for the item_popularity table as the offset store
    val itemPopularityKeyspace =
      system.settings.config
        .getString("akka.projection.cassandra.offset-store.keyspace")
    val itemPopularityRepository =
      new ItemPopularityRepositoryImpl(session, itemPopularityKeyspace)(
        system.executionContext
      ) (2)

    ItemPopularityProjection.init(system, itemPopularityRepository) (3)

@Override
public CompletionStage<GetItemPopularityResponse> getItemPopularity(GetItemPopularityRequest in) {
  return itemPopularityRepository
      .getItem(in.getItemId())
      .thenApply(
          maybePopularity -> {
            long popularity = maybePopularity.orElse(0L);
            return GetItemPopularityResponse.newBuilder().setPopularityCount(popularity).build();
          });
}

override def getItemPopularity(in: proto.GetItemPopularityRequest)
    : Future[proto.GetItemPopularityResponse] = {
  itemPopularityRepository.getItem(in.itemId).map {
    case Some(count) =>
      proto.GetItemPopularityResponse(in.itemId, count)
    case None =>
      proto.GetItemPopularityResponse(in.itemId, 0L)
  }
}