eventuate-client-java/src/main/java/io/eventuate/AggregateRepository.java
/**
   * Update the specified aggregate by processing a command and saving events
   * @param entityId the id of the aggregate to update
   * @param commandProvider the provider of the command to process
   * @param updateOptions options for updating
   * @return the updated and persisted aggregate
   */
Update the specified aggregate by processing a command and saving events
@param entityId the id of the aggregate to update
@param commandProvider the provider of the command to process
@param updateOptions options for updating
@return the updated and persisted aggregate
Update the specified aggregate by processing a command and saving events @param entityId the id of the aggregate to update @param commandProvider the provider of the command to process @param updateOptions options for updating @return the updated and persisted aggregate
public CompletableFuture<EntityWithIdAndVersion<T>> updateWithProvidedCommand(final String entityId, final Function<T, Optional<CT>> commandProvider, Optional<UpdateOptions> updateOptions) {
    return withRetry( () -> {
      CompletableFuture<LoadedEntityWithMetadata> eo = aggregateStore.find(clasz, entityId, updateOptions.map(uo -> new FindOptions().withTriggeringEvent(uo.getTriggeringEvent())))
              .handleAsync((tEntityWithMetadata, throwable) -> {
                if (throwable == null)
                  return new LoadedEntityWithMetadata(true, tEntityWithMetadata);
                else {
                  logger.debug("Exception finding aggregate", throwable);
                  Throwable unwrapped = CompletableFutureUtil.unwrap(throwable);
                  if (unwrapped instanceof DuplicateTriggeringEventException)
                    return new LoadedEntityWithMetadata(false, null);
                  else if (unwrapped instanceof RuntimeException)
                    throw (RuntimeException) unwrapped;
                  else if (throwable instanceof RuntimeException)
                    throw (RuntimeException) throwable;
                  else
                    throw new RuntimeException(throwable);
                }
              });
      return eo.thenCompose(loadedEntityWithMetadata -> {
        if (loadedEntityWithMetadata.success) {
          EntityWithMetadata<T> entityWithMetaData = loadedEntityWithMetadata.ewmd;
          final T aggregate = entityWithMetaData.getEntity();
          List<Event> events = commandProvider.apply(aggregate).map(aggregate::processCommand).orElse(Collections.emptyList());
          if (events.isEmpty()) {
            return CompletableFuture.completedFuture(entityWithMetaData.toEntityWithIdAndVersion());
          } else {
            return aggregateStore.update(clasz, entityWithMetaData.getEntityIdAndVersion(), events, updateOptions)
                    .thenApply(entityIdAndVersion -> {
                      Aggregates.applyEventsToMutableAggregate(aggregate, events);
                      return new EntityWithIdAndVersion<T>(entityIdAndVersion, aggregate);
                    })
                    .handle((r, t) -> {
                      if (t == null)
                        return r;
                      else {
                        logger.debug("Exception updating aggregate", t);
                        Throwable unwrapped = CompletableFutureUtil.unwrap(t);
                        if (unwrapped instanceof DuplicateTriggeringEventException)
                          return new EntityWithIdAndVersion<T>(entityWithMetaData.getEntityIdAndVersion(), aggregate);
                        else if (unwrapped instanceof RuntimeException)
                          throw (RuntimeException) unwrapped;
                        else if (t instanceof RuntimeException)
                          throw (RuntimeException) t;
                        else
                          throw new RuntimeException(unwrapped);
                      }
                    });
          }
        } else {
          return aggregateStore.find(clasz, entityId, Optional.empty()).thenApply(EntityWithMetadata::toEntityWithIdAndVersion);
        }
      });
    });
  }
public CompletableFuture < EntityWithIdAndVersion < T > > updateWithProvidedCommand ( final String entityId , final Function < T , Optional < CT > > commandProvider , Optional < UpdateOptions > updateOptions ) { return withRetry ( ( ) -> { CompletableFuture < LoadedEntityWithMetadata > eo = aggregateStore . find ( clasz , entityId , updateOptions . map ( uo -> new FindOptions ( ) . withTriggeringEvent ( uo . getTriggeringEvent ( ) ) ) ) . handleAsync ( ( tEntityWithMetadata , throwable ) -> { if ( throwable == null ) return new LoadedEntityWithMetadata ( true , tEntityWithMetadata ) ; else { logger . debug ( "Exception finding aggregate" , throwable ) ; Throwable unwrapped = CompletableFutureUtil . unwrap ( throwable ) ; if ( unwrapped instanceof DuplicateTriggeringEventException ) return new LoadedEntityWithMetadata ( false , null ) ; else if ( unwrapped instanceof RuntimeException ) throw ( RuntimeException ) unwrapped ; else if ( throwable instanceof RuntimeException ) throw ( RuntimeException ) throwable ; else throw new RuntimeException ( throwable ) ; } } ) ; return eo . thenCompose ( loadedEntityWithMetadata -> { if ( loadedEntityWithMetadata . success ) { EntityWithMetadata < T > entityWithMetaData = loadedEntityWithMetadata . ewmd ; final T aggregate = entityWithMetaData . getEntity ( ) ; List < Event > events = commandProvider . apply ( aggregate ) . map ( aggregate :: processCommand ) . orElse ( Collections . emptyList ( ) ) ; if ( events . isEmpty ( ) ) { return CompletableFuture . completedFuture ( entityWithMetaData . toEntityWithIdAndVersion ( ) ) ; } else { return aggregateStore . update ( clasz , entityWithMetaData . getEntityIdAndVersion ( ) , events , updateOptions ) . thenApply ( entityIdAndVersion -> { Aggregates . applyEventsToMutableAggregate ( aggregate , events ) ; return new EntityWithIdAndVersion < T > ( entityIdAndVersion , aggregate ) ; } ) . handle ( ( r , t ) -> { if ( t == null ) return r ; else { logger . debug ( "Exception updating aggregate" , t ) ; Throwable unwrapped = CompletableFutureUtil . unwrap ( t ) ; if ( unwrapped instanceof DuplicateTriggeringEventException ) return new EntityWithIdAndVersion < T > ( entityWithMetaData . getEntityIdAndVersion ( ) , aggregate ) ; else if ( unwrapped instanceof RuntimeException ) throw ( RuntimeException ) unwrapped ; else if ( t instanceof RuntimeException ) throw ( RuntimeException ) t ; else throw new RuntimeException ( unwrapped ) ; } } ) ; } } else { return aggregateStore . find ( clasz , entityId , Optional . empty ( ) ) . thenApply ( EntityWithMetadata :: toEntityWithIdAndVersion ) ; } } ) ; } ) ; }
Update the specified aggregate by processing a command and saving events
 @param entityId the id of the aggregate to update
 @param commandProvider the provider of the command to process
 @param updateOptions options for updating
 @return the updated and persisted aggregate
Update the specified aggregate by processing a command and saving events @param entityId the id of the aggregate to update @param commandProvider the provider of the command to process @param updateOptions options for updating @return the updated and persisted aggregate
// TODO - does this make sense?