diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublication.java b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublication.java index fe58ec86..8cc6a600 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublication.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublication.java @@ -18,6 +18,8 @@ package org.springframework.modulith.events.neo4j; import java.time.Instant; import java.util.UUID; +import org.springframework.lang.Nullable; + /** * The event publication entity definition. * @@ -32,15 +34,16 @@ class Neo4jEventPublication { public final Object event; public final String eventHash; - public Instant completionDate; + public @Nullable Instant completionDate; public Neo4jEventPublication(UUID identifier, Instant publicationDate, String listenerId, Object event, - String eventHash) { + String eventHash, @Nullable Instant completionDate) { this.identifier = identifier; this.publicationDate = publicationDate; this.listenerId = listenerId; this.event = event; this.eventHash = eventHash; + this.completionDate = completionDate; } } diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java index c68e75e8..d7f199c1 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java @@ -20,17 +20,20 @@ import static org.neo4j.cypherdsl.core.Cypher.*; import java.time.Instant; import java.time.ZoneOffset; import java.util.ArrayList; +import java.util.Collection; import java.util.List; import java.util.Map; import java.util.Objects; import java.util.Optional; import java.util.UUID; +import java.util.function.BiFunction; import java.util.function.Function; import org.neo4j.cypherdsl.core.Cypher; import org.neo4j.cypherdsl.core.Node; import org.neo4j.cypherdsl.core.ResultStatement; import org.neo4j.cypherdsl.core.Statement; +import org.neo4j.cypherdsl.core.StatementBuilder.OrderableOngoingReadingAndWithWithoutWhere; import org.neo4j.cypherdsl.core.renderer.Configuration; import org.neo4j.cypherdsl.core.renderer.Renderer; import org.neo4j.driver.Values; @@ -65,11 +68,14 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { private static final String PUBLICATION_DATE = "publicationDate"; private static final String COMPLETION_DATE = "completionDate"; + private static final Collection ALL_PROPERTIES = List.of(ID, EVENT_SERIALIZED, EVENT_HASH, EVENT_TYPE, + LISTENER_ID, PUBLICATION_DATE, COMPLETION_DATE); + private static final Node EVENT_PUBLICATION_NODE = node("Neo4jEventPublication") .named("neo4jEventPublication"); - private static final Node EVENT_PUBLICATION_COMPLETED_NODE = node("Neo4jEventPublicationCompleted") - .named("neo4jEventPublicationCompleted"); + private static final Node EVENT_PUBLICATION_ARCHIVE_NODE = node("Neo4jEventPublicationArchive") + .named("neo4jEventPublicationArchive"); private static final Statement INCOMPLETE_BY_EVENT_AND_TARGET_IDENTIFIER_STATEMENT = match(EVENT_PUBLICATION_NODE) .where(EVENT_PUBLICATION_NODE.property(EVENT_HASH).eq(parameter(EVENT_HASH))) @@ -114,6 +120,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .set(EVENT_PUBLICATION_NODE.property(EVENT_TYPE).to(parameter(EVENT_TYPE))) .set(EVENT_PUBLICATION_NODE.property(LISTENER_ID).to(parameter(LISTENER_ID))) .set(EVENT_PUBLICATION_NODE.property(PUBLICATION_DATE).to(parameter(PUBLICATION_DATE))) + .set(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).to(parameter(COMPLETION_DATE))) .build(); private static final Statement COMPLETE_STATEMENT = match(EVENT_PUBLICATION_NODE) @@ -123,29 +130,34 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .set(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).to(parameter(COMPLETION_DATE))) .build(); - private static final Statement COMPLETE_IN_ARCHIVE_BY_ID_STATEMENT = match(EVENT_PUBLICATION_NODE) + private static final Statement COMPLETE_IN_ARCHIVE_BY_ID_STATEMENT = applyProperties(match(EVENT_PUBLICATION_NODE) .where(EVENT_PUBLICATION_NODE.property(ID).eq(parameter(ID))) - .and(not(exists(match(EVENT_PUBLICATION_COMPLETED_NODE) - .where(EVENT_PUBLICATION_COMPLETED_NODE.property(ID).eq(parameter(ID))) + .and(not(exists(match(EVENT_PUBLICATION_ARCHIVE_NODE) + .where(EVENT_PUBLICATION_ARCHIVE_NODE.property(ID).eq(parameter(ID))) .returning(literalTrue()).build()))) - .with(EVENT_PUBLICATION_NODE) - .create(EVENT_PUBLICATION_COMPLETED_NODE) - .set(EVENT_PUBLICATION_COMPLETED_NODE.property(ID).to(EVENT_PUBLICATION_NODE.property(ID))) - .set(EVENT_PUBLICATION_COMPLETED_NODE.property(COMPLETION_DATE).to(parameter(COMPLETION_DATE))) - .build(); + .with(EVENT_PUBLICATION_NODE)); - private static final Statement COMPLETE_IN_ARCHIVE_BY_EVENT_AND_LISTENER_ID_STATEMENT = match(EVENT_PUBLICATION_NODE) - .where(EVENT_PUBLICATION_NODE.property(EVENT_HASH).eq(parameter(EVENT_HASH))) - .and(EVENT_PUBLICATION_NODE.property(LISTENER_ID).eq(parameter(LISTENER_ID))) - .and(not(exists(match(EVENT_PUBLICATION_COMPLETED_NODE) - .where(EVENT_PUBLICATION_COMPLETED_NODE.property(EVENT_HASH).eq(parameter(EVENT_HASH))) - .and(EVENT_PUBLICATION_COMPLETED_NODE.property(LISTENER_ID).eq(parameter(LISTENER_ID))) - .returning(literalTrue()).build()))) - .with(EVENT_PUBLICATION_NODE) - .create(EVENT_PUBLICATION_COMPLETED_NODE) - .set(EVENT_PUBLICATION_COMPLETED_NODE.property(ID).to(EVENT_PUBLICATION_NODE.property(ID))) - .set(EVENT_PUBLICATION_COMPLETED_NODE.property(COMPLETION_DATE).to(parameter(COMPLETION_DATE))) - .build(); + private static final Statement COMPLETE_IN_ARCHIVE_BY_EVENT_AND_LISTENER_ID_STATEMENT = applyProperties( + match(EVENT_PUBLICATION_NODE) + .where(EVENT_PUBLICATION_NODE.property(EVENT_HASH).eq(parameter(EVENT_HASH))) + .and(EVENT_PUBLICATION_NODE.property(LISTENER_ID).eq(parameter(LISTENER_ID))) + .and(not(exists(match(EVENT_PUBLICATION_ARCHIVE_NODE) + .where(EVENT_PUBLICATION_ARCHIVE_NODE.property(EVENT_HASH).eq(parameter(EVENT_HASH))) + .and(EVENT_PUBLICATION_ARCHIVE_NODE.property(LISTENER_ID).eq(parameter(LISTENER_ID))) + .returning(literalTrue()).build()))) + .with(EVENT_PUBLICATION_NODE)); + + private static Statement applyProperties(OrderableOngoingReadingAndWithWithoutWhere source) { + + var operations = ALL_PROPERTIES.stream() + .map(it -> EVENT_PUBLICATION_ARCHIVE_NODE.property(it).to(EVENT_PUBLICATION_NODE.property(it))) + .toList(); + + return source.create(EVENT_PUBLICATION_ARCHIVE_NODE) + .set(operations) + .set(EVENT_PUBLICATION_ARCHIVE_NODE.property(COMPLETION_DATE).to(parameter(COMPLETION_DATE))) + .build(); + } private static final Function COMPLETE_BY_ID_STATEMENT = node -> match(node) .where(node.property(ID).eq(parameter(ID))) @@ -168,6 +180,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { private final Renderer renderer; private final EventSerializer eventSerializer; private final CompletionMode completionMode; + private final Node completedNode; private final Statement deleteCompletedStatement; private final Statement deleteCompletedBeforeStatement; @@ -187,12 +200,14 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { this.eventSerializer = eventSerializer; this.completionMode = completionMode; - var archiveNode = completionMode == CompletionMode.ARCHIVE ? EVENT_PUBLICATION_COMPLETED_NODE : EVENT_PUBLICATION_NODE; + this.completedNode = completionMode == CompletionMode.ARCHIVE + ? EVENT_PUBLICATION_ARCHIVE_NODE + : EVENT_PUBLICATION_NODE; - this.deleteCompletedStatement = DELETE_COMPLETED_STATEMENT.apply(archiveNode); - this.deleteCompletedBeforeStatement = DELETE_COMPLETED_BEFORE_STATEMENT.apply(archiveNode); - this.completedByIdStatement = COMPLETE_BY_ID_STATEMENT.apply(archiveNode); - this.allCompletedStatement = ALL_COMPLETED_STATEMENT.apply(archiveNode); + this.deleteCompletedStatement = DELETE_COMPLETED_STATEMENT.apply(completedNode); + this.deleteCompletedBeforeStatement = DELETE_COMPLETED_BEFORE_STATEMENT.apply(completedNode); + this.completedByIdStatement = COMPLETE_BY_ID_STATEMENT.apply(completedNode); + this.allCompletedStatement = ALL_COMPLETED_STATEMENT.apply(completedNode); } /* @@ -219,7 +234,8 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { EVENT_HASH, eventHash, EVENT_TYPE, eventType, LISTENER_ID, listenerId, - PUBLICATION_DATE, Values.value(publicationDate.atOffset(ZoneOffset.UTC)))) + PUBLICATION_DATE, Values.value(publicationDate.atOffset(ZoneOffset.UTC)), + COMPLETION_DATE, Values.NULL)) .run(); return publication; @@ -249,6 +265,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .bind(identifier.getValue()).to(LISTENER_ID) .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) .run(); + neo4jClient.query(renderer.render(DELETE_BY_EVENT_AND_LISTENER_ID)) .bind(eventHash).to(EVENT_HASH) .bind(identifier.getValue()).to(LISTENER_ID) @@ -279,9 +296,10 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { } else if (completionMode == CompletionMode.ARCHIVE) { neo4jClient.query(renderer.render(COMPLETE_IN_ARCHIVE_BY_ID_STATEMENT)) - .bind("").to(ID) + .bind(Values.value(identifier.toString())).to(ID) .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) .run(); + deletePublications(List.of(identifier)); } else { @@ -304,7 +322,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { return List.copyOf(neo4jClient.query(renderer.render(INCOMPLETE_STATEMENT)) .fetchAs(TargetEventPublication.class) - .mappedBy(this::mapRecordToPublication) + .mappedBy(incompleteMapping()) .all()); } @@ -319,7 +337,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { return List.copyOf(neo4jClient.query(renderer.render(INCOMPLETE_PUBLISHED_BEFORE_STATEMENT)) .bind(Values.value(instant.atOffset(ZoneOffset.UTC))).to(PUBLICATION_DATE) .fetchAs(TargetEventPublication.class) - .mappedBy(this::mapRecordToPublication) + .mappedBy(incompleteMapping()) .all()); } @@ -338,7 +356,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { return neo4jClient.query(renderer.render(INCOMPLETE_BY_EVENT_AND_TARGET_IDENTIFIER_STATEMENT)) .bindAll(Map.of(EVENT_HASH, eventHash, LISTENER_ID, listenerId)) .fetchAs(TargetEventPublication.class) - .mappedBy(this::mapRecordToPublication) + .mappedBy(incompleteMapping()) .one(); } @@ -351,7 +369,7 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { return new ArrayList<>(neo4jClient.query(renderer.render(allCompletedStatement)) .fetchAs(TargetEventPublication.class) - .mappedBy(this::mapRecordToPublication) + .mappedBy(completeMapping()) .all()); } @@ -391,21 +409,31 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .run(); } - private Neo4jEventPublicationAdapter mapRecordToPublication(TypeSystem typeSystem, org.neo4j.driver.Record record) { + private BiFunction incompleteMapping() { + return (typeSystem, driverRecord) -> mapRecordToPublication(typeSystem, driverRecord, EVENT_PUBLICATION_NODE); + } - var publicationNode = record.get(EVENT_PUBLICATION_NODE.getRequiredSymbolicName().getValue()).asNode(); + private BiFunction completeMapping() { + return (typeSystem, driverRecord) -> mapRecordToPublication(typeSystem, driverRecord, completedNode); + } + + private Neo4jEventPublicationAdapter mapRecordToPublication(TypeSystem typeSystem, org.neo4j.driver.Record record, + Node node) { + + var publicationNode = record.get(node.getRequiredSymbolicName().getValue()).asNode(); var identifier = UUID.fromString(publicationNode.get(ID).asString()); var publicationDate = publicationNode.get(PUBLICATION_DATE).asZonedDateTime().toInstant(); var listenerId = publicationNode.get(LISTENER_ID).asString(); var eventSerialized = publicationNode.get(EVENT_SERIALIZED).asString(); var eventHash = publicationNode.get(EVENT_HASH).asString(); var eventType = publicationNode.get(EVENT_TYPE).asString(); + var completionDate = publicationNode.get(COMPLETION_DATE); try { var event = eventSerializer.deserialize(eventSerialized, Class.forName(eventType)); var publication = new Neo4jEventPublication(identifier, publicationDate, listenerId, event, - eventHash); + eventHash, completionDate.isNull() ? null : completionDate.asZonedDateTime().toInstant()); return new Neo4jEventPublicationAdapter(publication); diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java b/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java index a37f4b5e..ca46ffa7 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java @@ -42,9 +42,8 @@ import org.springframework.modulith.events.core.PublicationTargetIdentifier; import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.modulith.events.support.CompletionMode; import org.springframework.modulith.testapp.TestApplication; -import org.springframework.test.context.ContextConfiguration; import org.springframework.test.context.TestPropertySource; -import org.springframework.test.context.bean.override.mockito.MockitoBean; +import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.util.DigestUtils; import org.testcontainers.containers.Neo4jContainer; import org.testcontainers.junit.jupiter.Container; @@ -53,7 +52,12 @@ import org.testcontainers.utility.DockerImageName; /** * @author Gerrit Meier + * @author Cora Iberkleid + * @author Oliver Drotbohm */ +@Testcontainers(disabledWithoutDocker = true) +@SpringJUnitConfig(Neo4jEventPublicationRepositoryTest.Config.class) +@ImportAutoConfiguration(classes = Neo4jEventPublicationAutoConfiguration.class) class Neo4jEventPublicationRepositoryTest { static final PublicationTargetIdentifier TARGET_IDENTIFIER = PublicationTargetIdentifier.of("listener"); @@ -62,17 +66,12 @@ class Neo4jEventPublicationRepositoryTest { static final Neo4jContainer neo4jContainer = new Neo4jContainer<>(DockerImageName.parse("neo4j:5")) .withRandomPassword(); - @Import(TestApplication.class) - @ImportAutoConfiguration({ Neo4jEventPublicationAutoConfiguration.class }) - @Testcontainers(disabledWithoutDocker = true) - @ContextConfiguration(classes = Config.class) static abstract class TestBase { @Autowired Neo4jEventPublicationRepository repository; @Autowired Driver driver; @Autowired Environment environment; - - @MockitoBean EventSerializer eventSerializer; + @Autowired EventSerializer eventSerializer; CompletionMode completionMode; @@ -335,5 +334,10 @@ class Neo4jEventPublicationRepositoryTest { org.neo4j.cypherdsl.core.renderer.Configuration cypherDslConfiguration() { return org.neo4j.cypherdsl.core.renderer.Configuration.newConfig().withDialect(Dialect.NEO4J_5).build(); } + + @Bean + EventSerializer eventSerializer() { + return mock(EventSerializer.class); + } } }