EventGeneratorApp
This is a simulation of fake Event Sourced shopping carts. The details of this implementation is not important for understanding Projections. It’s needed for running the example.
Please look at the Apache Pekko reference documentation for Event Sourcing for how to implement real EventSourcedBehavior.
Add the EventGeneratorApp to your project:
sourcepackage docs.guide
import java.time.Instant
import scala.concurrent.duration._
import scala.util.Random
import org.apache.pekko
import pekko.actor.typed.ActorSystem
import pekko.actor.typed.Behavior
import pekko.actor.typed.scaladsl.Behaviors
import pekko.cluster.sharding.typed.scaladsl.ClusterSharding
import pekko.cluster.sharding.typed.scaladsl.Entity
import pekko.cluster.sharding.typed.scaladsl.EntityTypeKey
import pekko.cluster.typed.Cluster
import pekko.cluster.typed.Join
import pekko.persistence.typed.PersistenceId
import pekko.persistence.typed.scaladsl.Effect
import pekko.persistence.typed.scaladsl.EventSourcedBehavior
import pekko.stream.scaladsl.Sink
import pekko.stream.scaladsl.Source
import com.typesafe.config.ConfigFactory
/**
* Generate a shopping cart every 1 second and check it out. Each cart will contain a variety of `ItemAdded`,
* `ItemQuantityAdjusted` and `ItemRemoved` events preceding the the cart `Checkout` event.
*/
object EventGeneratorApp extends App {
import ShoppingCartEvents._
val Products = List("cat t-shirt", "pekko t-shirt", "skis", "bowling shoes")
val MaxQuantity = 5
val MaxItems = 3
val MaxItemsAdjusted = 3
val EntityKey: EntityTypeKey[Event] = EntityTypeKey[Event]("shopping-cart-event")
val config = ConfigFactory
.parseString("pekko.actor.provider = cluster")
.withFallback(ConfigFactory.load("guide-shopping-cart-app.conf"))
ActorSystem(Behaviors.setup[String] {
ctx =>
implicit val system = ctx.system
val cluster = Cluster(system)
cluster.manager ! Join(cluster.selfMember.address)
val sharding = ClusterSharding(system)
val _ = sharding.init(Entity(EntityKey) { entityCtx =>
cartBehavior(entityCtx.entityId, tagFactory(entityCtx.entityId))
})
Source
.tick(1.second, 1.second, "checkout")
.mapConcat {
case "checkout" =>
val cartId = java.util.UUID.randomUUID().toString.take(5)
val items = randomItems()
val itemEvents = (0 to items).flatMap {
_ =>
val itemId = Products(Random.nextInt(Products.size))
// add the item
val quantity = randomQuantity()
val itemAdded = ItemAdded(cartId, itemId, quantity)
// make up to `MaxItemAdjusted` adjustments to quantity of item
val adjustments = Random.nextInt(MaxItemsAdjusted)
val itemQuantityAdjusted = (0 to adjustments).foldLeft(Seq[ItemQuantityAdjusted]()) {
case (events, _) =>
val newQuantity = randomQuantity()
val oldQuantity =
if (events.isEmpty) itemAdded.quantity
else events.last.newQuantity
events :+ ItemQuantityAdjusted(cartId, itemId, newQuantity, oldQuantity)
}
// flip a coin to decide whether or not to remove the item
val itemRemoved =
if (Random.nextBoolean())
List(ItemRemoved(cartId, itemId, itemQuantityAdjusted.last.newQuantity))
else Nil
List(itemAdded) ++ itemQuantityAdjusted ++ itemRemoved
}
// checkout the cart and all its preceding item events
itemEvents :+ CheckedOut(cartId, Instant.now())
}
// send each event to the sharded entity represented by the event's cartId
.runWith(Sink.foreach(event => sharding.entityRefFor(EntityKey, event.cartId).ref.tell(event)))
Behaviors.empty
}, "EventGeneratorApp", config)
/**
* Random non-zero based quantity for `ItemAdded` and `ItemQuantityAdjusted` events
*/
def randomQuantity(): Int = Random.nextInt(MaxQuantity - 1) + 1
/**
* Random non-zero based count for how many `ItemAdded` events to generate
*/
def randomItems(): Int = Random.nextInt(MaxItems - 1) + 1
/**
* Choose a tag from `ShoppingCartTags` based on the entity id (cart id)
*/
def tagFactory(entityId: String): String =
if (args.contains("cluster")) {
val n = math.abs(entityId.hashCode % ShoppingCartTags.Tags.size)
val selectedTag = ShoppingCartTags.Tags(n)
selectedTag
} else ShoppingCartTags.Single
/**
* Construct an Actor that persists shopping cart events for a particular persistence id (cart id) and tag.
* This is not how real Event Sourced actors should be be implemented. Please look at
* https://pekko.apache.org/docs/pekko/current/typed/persistence.html for more information about `EventSourcedBehavior`.
*/
def cartBehavior(persistenceId: String, tag: String): Behavior[Event] =
Behaviors.setup { ctx =>
EventSourcedBehavior[Event, Event, List[Any]](
persistenceId = PersistenceId.ofUniqueId(persistenceId),
Nil,
(_, event) => {
ctx.log.info("id [{}] tag [{}] event: {}", persistenceId, tag, event)
Effect.persist(event)
},
(_, _) => Nil).withTagger(_ => Set(tag))
}
}sourcepackage jdocs.guide;
import com.typesafe.config.Config;
import com.typesafe.config.ConfigFactory;
import java.time.Duration;
import java.time.Instant;
import java.util.*;
import java.util.stream.Collectors;
import java.util.stream.IntStream;
import java.util.stream.Stream;
import org.apache.pekko.actor.typed.ActorSystem;
import org.apache.pekko.actor.typed.Behavior;
import org.apache.pekko.actor.typed.javadsl.Behaviors;
import org.apache.pekko.cluster.sharding.typed.javadsl.ClusterSharding;
import org.apache.pekko.cluster.sharding.typed.javadsl.Entity;
import org.apache.pekko.cluster.sharding.typed.javadsl.EntityTypeKey;
import org.apache.pekko.cluster.typed.Cluster;
import org.apache.pekko.cluster.typed.Join;
import org.apache.pekko.persistence.typed.PersistenceId;
import org.apache.pekko.persistence.typed.javadsl.CommandHandler;
import org.apache.pekko.persistence.typed.javadsl.EventHandler;
import org.apache.pekko.persistence.typed.javadsl.EventSourcedBehavior;
import org.apache.pekko.stream.javadsl.Sink;
import org.apache.pekko.stream.javadsl.Source;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
/**
* Generate a shopping cart every 1 second and check it out. Each cart will contain a variety of
* `ItemAdded`, `ItemQuantityAdjusted` and `ItemRemoved` events preceding the the cart `Checkout`
* event.
*/
public class EventGeneratorApp {
public static void main(String[] args) throws Exception {
Boolean clusterMode = (args.length > 0 && args[0].equals("cluster"));
Config config = config();
ActorSystem<String> system =
ActorSystem.create(Guardian.create(clusterMode), "EventGeneratorApp", config);
}
private static Config config() {
return ConfigFactory.parseString("pekko.actor.provider = cluster")
.withFallback(ConfigFactory.load("guide-shopping-cart-app.conf"));
}
}
class Guardian {
static final List<String> PRODUCTS =
Arrays.asList("cat t-shirt", "pekko t-shirt", "skis", "bowling shoes");
static final int MAX_QUANTITY = 5;
static final int MAX_ITEMS = 3;
static final int MAX_ITEMS_ADJUSTED = 3;
static final EntityTypeKey<ShoppingCartEvents.Event> ENTITY_KEY =
EntityTypeKey.create(ShoppingCartEvents.Event.class, "shopping-cart-event");
static Behavior<String> create(Boolean clusterMode) {
return Behaviors.setup(
context -> {
ActorSystem<Void> system = context.getSystem();
Cluster cluster = Cluster.get(system);
cluster.manager().tell(new Join(cluster.selfMember().address()));
ClusterSharding sharding = ClusterSharding.get(system);
sharding.init(
Entity.of(
ENTITY_KEY,
entityCtx -> {
PersistenceId persistenceId = PersistenceId.ofUniqueId(entityCtx.getEntityId());
String tag = tagFactory(entityCtx.getEntityId(), clusterMode);
return new CartPersistentBehavior(persistenceId, tag);
}));
Source.tick(Duration.ofSeconds(1L), Duration.ofSeconds(1L), "checkout")
.mapConcat(
checkout -> {
String cartId = UUID.randomUUID().toString().substring(0, 5);
int items = getRandomNumber(1, MAX_ITEMS);
Stream<ShoppingCartEvents.ItemEvent> itemEvents =
IntStream.range(0, items)
// .mapToObj(i -> Integer.valueOf(i)) // Java 8?
.boxed()
.flatMap(
i -> {
String itemId =
String.valueOf(getRandomNumber(0, PRODUCTS.size()));
ArrayList<ShoppingCartEvents.ItemEvent> events =
new ArrayList<>();
// add the item
int quantity = getRandomNumber(1, MAX_QUANTITY);
ShoppingCartEvents.ItemAdded itemAdded =
new ShoppingCartEvents.ItemAdded(cartId, itemId, quantity);
// make up to `MaxItemAdjusted` adjustments to quantity
// of item
int adjustments = getRandomNumber(0, MAX_ITEMS_ADJUSTED);
ArrayList<ShoppingCartEvents.ItemEvent> itemQuantityAdjusted =
new ArrayList<>();
for (int j = 0; j < adjustments; j++) {
int newQuantity = getRandomNumber(1, MAX_QUANTITY);
int oldQuantity = itemAdded.quantity;
if (!itemQuantityAdjusted.isEmpty()) {
oldQuantity =
((ShoppingCartEvents.ItemQuantityAdjusted)
itemQuantityAdjusted.get(
itemQuantityAdjusted.size() - 1))
.newQuantity;
}
itemQuantityAdjusted.add(
new ShoppingCartEvents.ItemQuantityAdjusted(
cartId, itemId, newQuantity, oldQuantity));
}
// flip a coin to decide whether or not to remove the
// item
ArrayList<ShoppingCartEvents.ItemEvent> itemRemoved =
new ArrayList<>();
if (Math.random() % 2 == 0) {
int oldQuantity =
((ShoppingCartEvents.ItemQuantityAdjusted)
itemQuantityAdjusted.get(
itemQuantityAdjusted.size() - 1))
.newQuantity;
itemRemoved.add(
new ShoppingCartEvents.ItemRemoved(
cartId, itemId, oldQuantity));
}
events.add(itemAdded);
events.addAll(itemQuantityAdjusted);
events.addAll(itemRemoved);
return events.stream();
});
// checkout the cart and all its preceding item events
return Stream.concat(
itemEvents,
Stream.of(new ShoppingCartEvents.CheckedOut(cartId, Instant.now())))
.collect(Collectors.toList());
})
// send each event to the sharded entity represented by the event's cartId
.runWith(
Sink.foreach(
event -> sharding.entityRefFor(ENTITY_KEY, event.getCartId()).tell(event)),
system);
return Behaviors.empty();
});
}
static int getRandomNumber(int min, int max) {
return (int) ((Math.random() * (max - min)) + min);
}
/** Choose a tag from `ShoppingCartTags` based on the entity id (cart id) */
static String tagFactory(String entityId, Boolean clusterMode) {
if (clusterMode) {
int n = Math.abs(entityId.hashCode() % ShoppingCartTags.TAGS.length);
String selectedTag = ShoppingCartTags.TAGS[n];
return selectedTag;
} else return ShoppingCartTags.SINGLE;
}
/**
* An Actor that persists shopping cart events for a particular persistence id (cart id) and tag.
* This is not how real Event Sourced actors should be be implemented. Please look at
* https://pekko.apache.org/docs/pekko/current/typed/persistence.html for more information about
* `EventSourcedBehavior`.
*/
static class CartPersistentBehavior
extends EventSourcedBehavior<
ShoppingCartEvents.Event, ShoppingCartEvents.Event, List<Object>> {
private final Logger log = LoggerFactory.getLogger(this.getClass());
private final String tag;
private final Set<String> tags;
public CartPersistentBehavior(PersistenceId persistenceId, String tag) {
super(persistenceId);
this.tag = tag;
this.tags = new HashSet<>(Collections.singletonList(tag));
}
@Override
public List<Object> emptyState() {
return new ArrayList<Object>();
}
@Override
public CommandHandler<ShoppingCartEvents.Event, ShoppingCartEvents.Event, List<Object>>
commandHandler() {
return (state, event) -> {
this.log.info("id [{}] tag [{}] event: {}", this.persistenceId().id(), this.tag, event);
return Effect().persist(event);
};
}
@Override
public EventHandler<List<Object>, ShoppingCartEvents.Event> eventHandler() {
return (state, event) -> state;
}
@Override
public Set<String> tagsFor(ShoppingCartEvents.Event event) {
return this.tags;
}
}
}
1.1.0