WIP optimisations

MAIF · Jul 10, 2024 · 2414065 · 2414065
1 parent 52b8cf1
commit 2414065
Show file tree

Hide file tree

Showing 13 changed files with 62 additions and 6 deletions.
diff --git a/demo/demo-in-memory/src/main/java/com/example/demo/Account.java b/demo/demo-in-memory/src/main/java/com/example/demo/Account.java
@@ -14,6 +14,11 @@ public Long sequenceNum() {
         return sequenceNum;
     }
 
+    @Override
+    public String entityId() {
+        return id;
+    }
+
     @Override
     public Account withSequenceNum(Long sequenceNum) {
         this.sequenceNum = sequenceNum;

diff --git a/demo/demo-postgres-kafka-reactive/src/main/java/com/example/demo/Account.java b/demo/demo-postgres-kafka-reactive/src/main/java/com/example/demo/Account.java
@@ -13,6 +13,11 @@ public String getId() {
         return id;
     }
 
+    @Override
+    public String entityId() {
+        return id;
+    }
+
     public BigDecimal getBalance() {
         return balance;
     }

diff --git a/demo/demo-postgres-kafka/src/main/java/com/example/demo/Account.java b/demo/demo-postgres-kafka/src/main/java/com/example/demo/Account.java
@@ -14,6 +14,11 @@ public Long sequenceNum() {
         return sequenceNum;
     }
 
+    @Override
+    public String entityId() {
+        return id;
+    }
+
     @Override
     public Account withSequenceNum(Long sequenceNum) {
         this.sequenceNum = sequenceNum;

diff --git a/sample/src/main/java/fr/maif/thoth/sample/state/Account.java b/sample/src/main/java/fr/maif/thoth/sample/state/Account.java
@@ -9,6 +9,11 @@ public class Account extends AbstractState<Account> {
     public BigDecimal balance;
     public long sequenceNum;
 
+    @Override
+    public String entityId() {
+        return id;
+    }
+
     public static class AccountBuilder{
         String id;
         BigDecimal balance;

diff --git a/thoth-core-akka/src/test/java/fr/maif/Helpers.java b/thoth-core-akka/src/test/java/fr/maif/Helpers.java
@@ -354,6 +354,11 @@ public Viking(String id, String name, Long sequenceNum) {
             this.sequenceNum = sequenceNum;
         }
 
+        @Override
+        public String entityId() {
+            return id;
+        }
+
         @Override
         public Long sequenceNum() {
             return sequenceNum;

diff --git a/thoth-core-reactor/src/test/java/fr/maif/Helpers.java b/thoth-core-reactor/src/test/java/fr/maif/Helpers.java
@@ -415,6 +415,11 @@ public Viking(String id, String name, Integer age, Long sequenceNum) {
             this.sequenceNum = sequenceNum;
         }
 
+        @Override
+        public String entityId() {
+            return id;
+        }
+
         @Override
         public Long sequenceNum() {
             return sequenceNum;

diff --git a/thoth-core/src/main/java/fr/maif/eventsourcing/AbstractState.java b/thoth-core/src/main/java/fr/maif/eventsourcing/AbstractState.java
@@ -1,6 +1,6 @@
 package fr.maif.eventsourcing;
 
-public class AbstractState<T> implements State<T> {
+public abstract class AbstractState<T> implements State<T> {
     protected long sequenceNum;
 
     @Override

diff --git a/thoth-core/src/main/java/fr/maif/eventsourcing/AggregateStore.java b/thoth-core/src/main/java/fr/maif/eventsourcing/AggregateStore.java
@@ -3,6 +3,7 @@
 import fr.maif.concurrent.CompletionStages;
 import io.vavr.Tuple;
 import io.vavr.Tuple0;
+import io.vavr.collection.HashMap;
 import io.vavr.collection.List;
 import io.vavr.collection.Map;
 import io.vavr.control.Option;
@@ -13,7 +14,10 @@ public interface AggregateStore<S extends State<S>, Id, TxCtx> {
 
     CompletionStage<Option<S>> getAggregate(Id entityId);
 
-    CompletionStage<Map<String, Option<S>>> getAggregates(TxCtx ctx, List<String> entityIds);
+    default CompletionStage<Map<Id, Option<S>>> getAggregates(TxCtx ctx, List<Id> entityIds) {
+        return CompletionStages.traverse(entityIds, id -> getAggregate(ctx, id).thenApply(agg -> Tuple.of(id, agg)))
+                .thenApply(HashMap::ofEntries);
+    }
 
     CompletionStage<Option<S>> getAggregate(TxCtx ctx, Id entityId);
 

diff --git a/thoth-core/src/test/java/fr/maif/Helpers.java b/thoth-core/src/test/java/fr/maif/Helpers.java
@@ -345,6 +345,11 @@ public Long sequenceNum() {
             return sequenceNum;
         }
 
+        @Override
+        public String entityId() {
+            return id;
+        }
+
         @Override
         public Viking withSequenceNum(Long sequenceNum) {
             this.sequenceNum = sequenceNum;

diff --git a/thoth-jooq-akka/src/main/java/fr/maif/eventsourcing/ReactivePostgresEventStore.java b/thoth-jooq-akka/src/main/java/fr/maif/eventsourcing/ReactivePostgresEventStore.java
@@ -21,6 +21,7 @@
 import io.vavr.Tuple0;
 import io.vavr.collection.List;
 import io.vavr.collection.Seq;
+import io.vavr.collection.Traversable;
 import io.vavr.concurrent.Future;
 import io.vavr.control.Either;
 import io.vavr.control.Option;
@@ -228,7 +229,10 @@ public Publisher<EventEnvelope<E, Meta, Context>> loadEventsByQuery(PgAsyncTrans
                 query.userId().map(USER_ID::eq),
                 query.published().map(PUBLISHED::eq),
                 query.sequenceTo().map(SEQUENCE_NUM::le),
-                query.sequenceFrom().map(SEQUENCE_NUM::ge)
+                query.sequenceFrom().map(SEQUENCE_NUM::ge),
+                Option.of(query.idsAndSequences()).filter(Traversable::nonEmpty).map(l ->
+                        l.map(t -> SEQUENCE_NUM.ge(t._2).and(ENTITY_ID.eq(t._1))).reduce(Condition::or)
+                )
         ).flatMap(identity());
 
         return Source.fromPublisher(tx.stream(500, dsl -> {

diff --git a/thoth-jooq-reactor/src/main/java/fr/maif/eventsourcing/ReactivePostgresEventStore.java b/thoth-jooq-reactor/src/main/java/fr/maif/eventsourcing/ReactivePostgresEventStore.java
@@ -15,6 +15,7 @@
 import io.vavr.Tuple0;
 import io.vavr.collection.List;
 import io.vavr.collection.Seq;
+import io.vavr.collection.Traversable;
 import io.vavr.control.Either;
 import io.vavr.control.Option;
 import io.vavr.control.Try;
@@ -260,7 +261,10 @@ public Publisher<EventEnvelope<E, Meta, Context>> loadEventsByQuery(Tx tx, Query
                 query.userId().map(USER_ID::eq),
                 query.published().map(PUBLISHED::eq),
                 query.sequenceTo().map(SEQUENCE_NUM::le),
-                query.sequenceFrom().map(SEQUENCE_NUM::ge)
+                query.sequenceFrom().map(SEQUENCE_NUM::ge),
+                Option.of(query.idsAndSequences()).filter(Traversable::nonEmpty).map(l ->
+                    l.map(t -> SEQUENCE_NUM.ge(t._2).and(ENTITY_ID.eq(t._1))).reduce(Condition::or)
+                )
         ).flatMap(identity());
 
         return Flux.from(tx.stream(500, dsl -> {

diff --git a/thoth-jooq/src/main/java/fr/maif/eventsourcing/impl/PostgresEventStore.java b/thoth-jooq/src/main/java/fr/maif/eventsourcing/impl/PostgresEventStore.java
@@ -308,7 +308,10 @@ public Publisher<EventEnvelope<E, Meta, Context>> loadEventsByQueryWithOptions(C
                 query.userId().map(d -> field(" user_id").eq(d)),
                 query.published().map(d -> field(" published").eq(d)),
                 query.sequenceTo().map(d -> field(" sequence_num").lessOrEqual(d)),
-                query.sequenceFrom().map(d -> field(" sequence_num").greaterOrEqual(d))
+                query.sequenceFrom().map(d -> field(" sequence_num").greaterOrEqual(d)),
+                Option.of(query.idsAndSequences()).filter(Traversable::nonEmpty).map(l ->
+                        l.map(t -> field(" sequence_num").greaterOrEqual(t._2).and(field(" entity_id").eq(t._1))).reduce(Condition::or)
+                )
         ).flatMap(identity());
 
         var tmpJooqQuery = DSL.using(tx)

diff --git a/thoth-tck/src/main/java/fr/maif/eventsourcing/datastore/TestState.java b/thoth-tck/src/main/java/fr/maif/eventsourcing/datastore/TestState.java
@@ -6,9 +6,15 @@ public class TestState extends AbstractState<TestState> {
     public final String id;
     public final int count;
 
-
     public TestState(String id, int count) {
         this.id = id;
         this.count = count;
     }
+
+    @Override
+    public String entityId() {
+        return id;
+    }
+
+
 }