Developer best practices

public class CustomerStore {

  public CompletionStage<Optional<Customer>> getById(String customerId) {
  }

  public CompletionStage<Done> save(String customerId, Customer customer) {
  }
}

javaCopied!

@ComponentId("customer-store-updater")
@Consume.FromEventSourcedEntity(CustomerEntity.class)
public class CustomerStoreUpdater extends Consumer {

  private final CustomerStore customerStore;

  public CustomerStoreUpdater(CustomerStore customerStore) {
    this.customerStore = customerStore;
  }

  public Effect onEffect(CustomerEvent event) { (1)
    var customerId = messageContext().eventSubject().get();
    return switch (event) {
      case CustomerCreated created ->
        effects().asyncDone(customerStore.save(customerId, new Customer(created.email(), created.name(), created.address())));

      case NameChanged nameChanged ->
        effects().asyncDone(customerStore.getById(customerId).thenCompose(customer -> {
          if (customer.isPresent()) {
            return customerStore.save(customerId, customer.get().withName(nameChanged.newName()));
          } else {
            throw new IllegalStateException("Customer not found: " + customerId);
          }
        }));

      case AddressChanged addressChanged ->
        effects().asyncDone(customerStore.getById(customerId).thenCompose(customer -> {
          if (customer.isPresent()) {
            return customerStore.save(customerId, customer.get().withAddress(addressChanged.address()));
          } else {
            throw new IllegalStateException("Customer not found: " + customerId);
          }
        }));
    };
  }
}

javaCopied!

@ComponentId("counter-by-value")
public class CounterByValueView extends View {

  public record CounterByValue(String name, int value) {
  }

  @Consume.FromEventSourcedEntity(CounterEntity.class)
  public static class CounterByValueUpdater extends TableUpdater<CounterByValue> {
    public Effect<CounterByValue> onEvent(CounterEvent counterEvent) {
      var name = updateContext().eventSubject().get();
      var currentRow = rowState();
      var currentValue = Optional.ofNullable(currentRow).map(CounterByValue::value).orElse(0);
      return switch (counterEvent) {
        case ValueIncreased increased -> effects().updateRow(
          new CounterByValue(name, currentValue + increased.value())); (1)
        case ValueMultiplied multiplied -> effects().updateRow(
          new CounterByValue(name, currentValue * multiplied.multiplier())); (2)
      };
    }
  }
}

javaCopied!

public sealed interface CounterEvent {

  record ValueMultiplied(int multiplier,
                         int updatedValue) (1)
    implements CounterEvent {
  }
}

javaCopied!

  @Consume.FromEventSourcedEntity(CounterEntity.class)
  public static class CounterByValueUpdater extends TableUpdater<CounterByValue> {
    public Effect<CounterByValue> onEvent(CounterEvent counterEvent) {
      var name = updateContext().eventSubject().get();
      return switch (counterEvent) {
        case ValueIncreased increased -> effects().updateRow(
          new CounterByValue(name, increased.updatedValue())); (1)
        case ValueMultiplied multiplied -> effects().updateRow(
          new CounterByValue(name, multiplied.updatedValue())); (1)
      };
    }
  }
}

javaCopied!

public class CounterStore{

  public record CounterEntry(String counterId, int value, long seqNum) { (1)
  }

  private Map<String, CounterEntry> store = new ConcurrentHashMap<>();


  public CompletionStage<Optional<CounterEntry>> getById(String counterId) {
    return completedFuture(Optional.ofNullable(store.get(counterId)));
  }

  public CompletionStage<Done> save(CounterEntry counterEntry) {
    return completedFuture(store.put(counterEntry.counterId(), counterEntry)).thenApply(__ -> done());
  }

  public CompletionStage<Collection<CounterEntry>> getAll() {
    return completedFuture(store.values());
  }
}

javaCopied!

@ComponentId("counter-store-updater")
@Consume.FromEventSourcedEntity(CounterEntity.class)
public class CounterStoreUpdater extends Consumer {

  private final CounterStore counterStore;

  public Effect onEvent(CounterEvent counterEvent) {
    var counterId = messageContext().eventSubject().get();
    var newSeqNum = messageContext().metadata().asCloudEvent().sequence();

    return effects().asyncEffect(
      counterStore.getById(counterId) (1)
        .thenApply(counterEntry -> {
          var currentSeqNum = counterEntry.map(CounterEntry::seqNum).orElse(0L);
          if (!newSeqNum.isPresent()) { (2)
            // missing sequence number, can't deduplicate
            return processEvent(counterEvent, counterEntry, 0L);
          } else {
            if (newSeqNum.get() <= currentSeqNum) {
              //duplicate, can be ignored
              return effects().ignore(); (3)
            } else {
              // not a duplicate
              return processEvent(counterEvent, counterEntry, newSeqNum.get()); (4)
            }
          }
        })
    );
  }

  private Effect processEvent(CounterEvent counterEvent, Optional<CounterEntry> currentEntry, Long seqNum) {
    var counterId = messageContext().eventSubject().get();
    var currentValue = currentEntry.map(CounterEntry::value).orElse(0);
    return switch (counterEvent) {
      case ValueIncreased increased -> {
        var updatedEntry = new CounterEntry(counterId, currentValue + increased.value(), seqNum);
        yield effects().asyncDone(counterStore.save(updatedEntry)); (5)
      }
      case ValueMultiplied multiplied -> {
        var updatedEntry = new CounterEntry(counterId, currentValue * multiplied.multiplier(), seqNum);
        yield effects().asyncDone(counterStore.save(updatedEntry)); (5)
      }
    };
  }
}

javaCopied!

public Effect handle(Event event) {
  var entityId = messageContext().eventSubject().get();
  var sequenceNumber = messageContext().metadata().asCloudEvent().sequence().get();
  var token = UUID.nameUUIDFromBytes((entityId + sequenceNumber).getBytes()); (1)
  return effects().asyncDone(
    someService.doSomething(event, token.toString()));
}

javaCopied!