From b189dbc7254e9565565ec00c7077e9c55e30a466 Mon Sep 17 00:00:00 2001 From: Oliver Drotbohm Date: Thu, 24 Oct 2024 21:11:15 +0200 Subject: [PATCH] GH-806 - Polish Neo4j archiving mode. We now explicitly set the completion date property of newly created event publications to null. The copying data manipulation statements now properly copy all properties of the publication notes into the archive. --- .../events/neo4j/Neo4jEventPublication.java | 7 +- .../Neo4jEventPublicationRepository.java | 100 +++++++++++------- .../Neo4jEventPublicationRepositoryTest.java | 20 ++-- 3 files changed, 81 insertions(+), 46 deletions(-) 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); + } } }