Update readme and code cleaning

common-persistence-api/src/main/java/com/mz/reactor/ddd/reactorddd/persistance/aggregate/impl/AggregateFacadeImpl.java

@@ -44,7 +44,7 @@ public class AggregateFacadeImpl<A, C extends Command, S> implements AggregateFa
   public Mono<? extends DomainEvent> executeReturnEvent(C command, String aggregateID, Class<? extends DomainEvent> eventType) {
     var result = aggregateRepository.execute(command, new Id(aggregateID));
     return result.flatMap(cr -> processResult(aggregateID, eventType, cr))
-        .doOnError(error -> log.error("execute -> event type: "+eventType, error));
+        .doOnError(error -> log.error("execute -> event type: " + eventType, error));
   }
 
   @Override

common-persistence-api/src/main/java/com/mz/reactor/ddd/reactorddd/persistance/aggregate/impl/AggregateRepositoryImpl.java

@@ -23,87 +23,88 @@ import static java.util.stream.Collectors.toList;
 
 public class AggregateRepositoryImpl<A, C extends Command, S> implements AggregateRepository<A, C, S> {
 
-    private final AtomicReference<Map<Id, List<? extends DomainEvent>>> eventSource = new AtomicReference<>(new HashMap<>());
+  private final AtomicReference<Map<Id, List<? extends DomainEvent>>> eventSource = new AtomicReference<>(new HashMap<>());
 
-    private final Cache<Id, AggregateActor<A, C>> cache = CacheBuilder.newBuilder()
-            .expireAfterAccess(Duration.ofMinutes(10))
-            .removalListener((RemovalListener<Id, AggregateActor<A, C>>) notification -> notification.getValue().onDestroy())
-            .build();
+  private final Cache<Id, AggregateActor<A, C>> cache = CacheBuilder.newBuilder()
+      .expireAfterAccess(Duration.ofMinutes(10))
+      .removalListener((RemovalListener<Id, AggregateActor<A, C>>) notification -> notification.getValue().onDestroy())
+      .build();
 
 
-    private final CommandHandler<A, C> commandHandler;
+  private final CommandHandler<A, C> commandHandler;
 
-    private final EventHandler<A> eventHandler;
+  private final EventHandler<A> eventHandler;
 
-    private final Function<Id, A> aggregateFactory;
+  private final Function<Id, A> aggregateFactory;
 
-    private final Function<A, S> stateFactory;
+  private final Function<A, S> stateFactory;
 
-    public AggregateRepositoryImpl(
-            CommandHandler<A, C> commandHandler,
-            EventHandler<A> eventHandler,
-            Function<Id, A> aggregateFactory,
-            Function<A, S> stateFactory
-    ) {
-        this.commandHandler = commandHandler;
-        this.eventHandler = eventHandler;
-        this.aggregateFactory = aggregateFactory;
-        this.stateFactory = stateFactory;
+  public AggregateRepositoryImpl(
+      CommandHandler<A, C> commandHandler,
+      EventHandler<A> eventHandler,
+      Function<Id, A> aggregateFactory,
+      Function<A, S> stateFactory
+  ) {
+    this.commandHandler = commandHandler;
+    this.eventHandler = eventHandler;
+    this.aggregateFactory = aggregateFactory;
+    this.stateFactory = stateFactory;
+  }
+
+  private List<? extends DomainEvent> persistAll(Id id, List<? extends DomainEvent> events) {
+    eventSource.updateAndGet(esMap -> {
+      var eventsToStore = Optional.ofNullable(esMap.get(id))
+          .map(es -> Stream.concat(es.stream(), events.stream())
+              .sorted(Comparator.comparing(DomainEvent::createdAt))
+              .collect(toList()))
+          .orElse((List<DomainEvent>) events);
+      esMap.put(id, eventsToStore);
+      return esMap;
+    });
+    return events;
+  }
+
+  private Mono<AggregateActor<A, C>> getAggregate(Id id) {
+    return Mono.just(id)
+        .map(this::getFromCache);
+  }
+
+  private AggregateActor<A, C> getFromCache(Id id) {
+    try {
+      return cache.get(id, () ->
+          new AggregateActorImpl<>(
+              id,
+              commandHandler,
+              eventHandler,
+              aggregateFactory,
+              Optional.ofNullable(eventSource.get().get(id)).orElseGet(List::of),
+              this::persistAll
+          )
+      );
+    } catch (Exception e) {
+      throw new RuntimeException(e);
     }
+  }
 
-    private List<? extends DomainEvent> persistAll(Id id, List<? extends DomainEvent> events) {
-        eventSource.updateAndGet(esMap -> {
-            var eventsToStore = Optional.ofNullable(esMap.get(id))
-                    .map(es -> Stream.concat(es.stream(), events.stream())
-                            .sorted(Comparator.comparing(DomainEvent::createdAt))
-                            .collect(toList()))
-                    .orElse((List<DomainEvent>) events);
-            esMap.put(id, eventsToStore);
-            return esMap;
-        });
-        return events;
-    }
+  @Override
+  public Mono<CommandResult> execute(C cmd, Id aggregateId) {
+    return getAggregate(aggregateId)
+        .flatMap(a -> a.execute(cmd));
+  }
 
-    private Mono<AggregateActor<A, C>> getAggregate(Id id) {
-        return Mono.just(id)
-                .map(this::getFromCache);
-    }
+  @Override
+  public Mono<S> findById(Id id) {
+    return getAggregate(id).flatMap(a -> a.getState(this.stateFactory));
+  }
 
-    private AggregateActor<A, C> getFromCache(Id id) {
-        try {
-            return cache.get(id, () ->
-                    new AggregateActorImpl<A, C>(
-                            id,
-                            commandHandler,
-                            eventHandler,
-                            aggregateFactory,
-                            Optional.ofNullable(eventSource.get().get(id)).orElseGet(List::of),
-                            this::persistAll
-                    )
-            );
-        } catch (Exception e) {
-            throw new RuntimeException(e);
-        }
-    }
-
-    @Override
-    public Mono<CommandResult> execute(C cmd, Id aggregateId) {
-        return getAggregate(aggregateId)
-                .flatMap(a -> a.execute(cmd));
-    }
-
-    @Override
-    public Mono<S> findById(Id id) {
-        return getAggregate(id).flatMap(a -> a.getState(this.stateFactory));
-    }
-
-    @Override
-    public Mono<S> findIfExists(Id id) {
-        return Mono.just(id)
-                .flatMap(i -> Optional.ofNullable(cache.getIfPresent(i))
-                        .map(a -> a.getState(this.stateFactory))
-                        .orElseGet(Mono::empty)
-                );
-    }
+  @Override
+  public Mono<S> findIfExists(Id id) {
+    return Mono.just(id)
+        .flatMap(
+            i -> Optional.ofNullable(cache.getIfPresent(i))
+                .map(a -> a.getState(this.stateFactory))
+                .orElseGet(Mono::empty)
+        );
+  }
 
 }

common-persistence-api/src/main/java/com/mz/reactor/ddd/reactorddd/persistance/query/Query.java

@@ -0,0 +1,20 @@
+package com.mz.reactor.ddd.reactorddd.persistance.query;
+
+import com.mz.reactor.ddd.common.api.view.DomainView;
+import com.mz.reactor.ddd.reactorddd.persistance.query.impl.QueryImpl;
+import com.mz.reactor.ddd.reactorddd.persistance.view.impl.ViewRepository;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Supplier;
+
+public interface Query<S extends DomainView> {
+
+  Mono<S> findById(String id);
+
+  Flux<S> getAll();
+
+  static <S extends DomainView> Query<S> of(ViewRepository<S> viewRepository, Supplier<Flux<S>> documentStream) {
+    return new QueryImpl<>(viewRepository, documentStream);
+  }
+}

common-persistence-api/src/main/java/com/mz/reactor/ddd/reactorddd/persistance/query/impl/QueryImpl.java

@@ -0,0 +1,39 @@
+package com.mz.reactor.ddd.reactorddd.persistance.query.impl;
+
+import com.mz.reactor.ddd.common.api.view.DomainView;
+import com.mz.reactor.ddd.reactorddd.persistance.query.Query;
+import com.mz.reactor.ddd.reactorddd.persistance.view.impl.ViewRepository;
+import org.springframework.lang.NonNull;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import java.util.function.Predicate;
+import java.util.function.Supplier;
+
+import static java.util.Objects.requireNonNull;
+
+public class QueryImpl<S extends DomainView> implements Query<S> {
+
+  private final ViewRepository<S> repository;
+
+  private final Predicate<S> getAll = v -> true;
+
+  public QueryImpl(
+      ViewRepository<S> repository,
+      Supplier<Flux<S>> documentStream
+  ) {
+    this.repository = requireNonNull(repository, "repository is required");
+    requireNonNull(documentStream, "documentStream is required").get()
+        .subscribe(repository::addView);
+  }
+
+  @Override
+  public Mono<S> findById(@NonNull String id) {
+    return repository.findBy(view -> view.id().equals(id));
+  }
+
+  @Override
+  public Flux<S> getAll() {
+    return repository.findAllBy(getAll);
+  }
+}