WIP-CRUD on top of Event Sourcing

java-support/src/main/scala/io/cloudstate/javasupport/impl/eventsourced/EventSourcedImpl.scala

@@ -131,9 +131,10 @@ final class EventSourcedImpl(_system: ActorSystem,
       .map(_.message)
       .scan[(Long, Option[EventSourcedStreamOut.Message])]((startingSequenceNumber, None)) {
         case (_, InEvent(event)) =>
-          val context = new EventContextImpl(entityId, event.sequence)
-          val ev = ScalaPbAny.toJavaProto(event.payload.get) // FIXME empty?
-          handler.handleEvent(ev, context)
+          // No needed here because the UserFunction does not deal with events
+          //val context = new EventContextImpl(entityId, event.sequence)
+          //val ev = ScalaPbAny.toJavaProto(event.payload.get) // FIXME empty?
+          //handler.handleEvent(ev, context)
           (event.sequence, None)
         case ((sequence, _), InCommand(command)) =>
           if (entityId != command.entityId)
@@ -226,9 +227,12 @@ final class EventSourcedImpl(_system: ActorSystem,
 
     override def emit(event: AnyRef): Unit = {
       checkActive()
+      //TODO: we need some kind of abstraction here between Event Sourcing and CRUD
+      //TODO: investigate to do compression here because it is the whole object graph which can be big
       val encoded = anySupport.encodeScala(event)
       val nextSequenceNumber = sequenceNumber + events.size + 1
-      handler.handleEvent(ScalaPbAny.toJavaProto(encoded), new EventContextImpl(entityId, nextSequenceNumber))
+      // No needed here because the UserFunction does not deal with events
+      //handler.handleEvent(ScalaPbAny.toJavaProto(encoded), new EventContextImpl(entityId, nextSequenceNumber))
       events :+= encoded
       performSnapshot = (snapshotEvery > 0) && (performSnapshot || (nextSequenceNumber % snapshotEvery == 0))
     }

proxy/core/src/main/scala/io/cloudstate/proxy/eventsourced/EventSourcedEntity.scala

@@ -306,7 +306,10 @@ final class EventSourcedEntity(configuration: EventSourcedEntity.Configuration,
       throw error
 
     case SaveSnapshotSuccess(metadata) =>
-    // Nothing to do
+      // TODO: is it he right place delete the oldest snapshots and events?
+      // TODO ideally we want a one to one mapping between event and snapshot.
+      deleteSnapshots(SnapshotSelectionCriteria().copy(maxSequenceNr = metadata.sequenceNr - 1))
+      deleteMessages(metadata.sequenceNr - 1)
 
     case SaveSnapshotFailure(metadata, cause) =>
       log.error("Error saving snapshot", cause)
@@ -349,7 +352,6 @@ final class EventSourcedEntity(configuration: EventSourcedEntity.Configuration,
 
     case event: pbAny =>
       maybeInit(None)
-      relay ! EventSourcedStreamIn(EventSourcedStreamIn.Message.Event(EventSourcedEvent(lastSequenceNr, Some(event))))
   }
 
   private def reportDatabaseOperationStarted(): Unit =

samples/java-shopping-cart/src/main/java/io/cloudstate/samples/shoppingcart/ShoppingCartCrudEntity.java

@@ -0,0 +1,107 @@
+package io.cloudstate.samples.shoppingcart;
+
+import com.example.shoppingcart.Shoppingcart;
+import com.example.shoppingcart.persistence.Domain;
+import com.google.protobuf.Empty;
+import io.cloudstate.javasupport.EntityId;
+import io.cloudstate.javasupport.eventsourced.CommandContext;
+import io.cloudstate.javasupport.eventsourced.CommandHandler;
+import io.cloudstate.javasupport.eventsourced.EventSourcedEntity;
+import io.cloudstate.javasupport.eventsourced.Snapshot;
+import io.cloudstate.javasupport.eventsourced.SnapshotHandler;
+
+import java.util.LinkedHashMap;
+import java.util.List;
+import java.util.Map;
+import java.util.stream.Collectors;
+
+/** An event sourced entity. */
+@EventSourcedEntity(snapshotEvery = 1)
+public class ShoppingCartCrudEntity {
+  private final String entityId;
+  private final Map<String, Shoppingcart.LineItem> cart = new LinkedHashMap<>();
+
+  public ShoppingCartCrudEntity(@EntityId String entityId) {
+    this.entityId = entityId;
+  }
+
+  @Snapshot
+  public Domain.Cart snapshot() {
+    return Domain.Cart.newBuilder()
+        .addAllItems(cart.values().stream().map(this::convert).collect(Collectors.toList()))
+        .build();
+  }
+
+  @SnapshotHandler
+  public void handleSnapshot(Domain.Cart cart) {
+    this.cart.clear();
+    for (Domain.LineItem item : cart.getItemsList()) {
+      this.cart.put(item.getProductId(), convert(item));
+    }
+  }
+
+  @CommandHandler
+  public Shoppingcart.Cart getCart() {
+    return Shoppingcart.Cart.newBuilder().addAllItems(cart.values()).build();
+  }
+
+  @CommandHandler
+  public Empty addItem(Shoppingcart.AddLineItem item, CommandContext ctx) {
+    if (item.getQuantity() <= 0) {
+      ctx.fail("Cannot add negative quantity of to lineItem" + item.getProductId());
+    }
+
+    Domain.LineItem lineItem =
+        Domain.LineItem.newBuilder()
+            .setProductId(item.getProductId())
+            .setName(item.getName())
+            .setQuantity(quantity(item))
+            .build();
+
+    List<Domain.LineItem> lineItems =
+        cart.values().stream()
+            .map(this::convert)
+            .filter(someItem -> !someItem.getProductId().equals(item.getProductId()))
+            .collect(Collectors.toList());
+
+    ctx.emit(Domain.Cart.newBuilder().addAllItems(lineItems).addItems(lineItem).build());
+    return Empty.getDefaultInstance();
+  }
+
+  @CommandHandler
+  public Empty removeItem(Shoppingcart.RemoveLineItem item, CommandContext ctx) {
+    if (!cart.containsKey(item.getProductId())) {
+      ctx.fail("Cannot remove item " + item.getProductId() + " because it is not in the cart.");
+    }
+
+    List<Domain.LineItem> lineItems =
+        cart.values().stream()
+            .map(this::convert)
+            .filter(someItem -> someItem.getProductId().equals(item.getProductId()))
+            .collect(Collectors.toList());
+
+    ctx.emit(Domain.Cart.newBuilder().addAllItems(lineItems).build());
+    return Empty.getDefaultInstance();
+  }
+
+  private Shoppingcart.LineItem convert(Domain.LineItem item) {
+    return Shoppingcart.LineItem.newBuilder()
+        .setProductId(item.getProductId())
+        .setName(item.getName())
+        .setQuantity(item.getQuantity())
+        .build();
+  }
+
+  private Domain.LineItem convert(Shoppingcart.LineItem item) {
+    return Domain.LineItem.newBuilder()
+        .setProductId(item.getProductId())
+        .setName(item.getName())
+        .setQuantity(item.getQuantity())
+        .build();
+  }
+
+  private int quantity(Shoppingcart.AddLineItem item) {
+    Shoppingcart.LineItem lineItem = cart.get(item.getProductId());
+    return lineItem == null ? item.getQuantity() : lineItem.getQuantity() + item.getQuantity();
+  }
+}