Spring/spring-modulith
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-806 - Polish Neo4j archiving mode.

Browse Source 
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.
This commit is contained in:
Oliver Drotbohm
2024-10-24 21:11:15 +02:00
parent 4132ad3a38
commit b189dbc725
3 changed files with 81 additions and 46 deletions
Download Patch File Download Diff File Expand all files Collapse all files

7
spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublication.java
View File

@@ -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;
}
}

100
spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepository.java
View File

@@ -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<String> 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<Node, Statement> 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<TypeSystem, org.neo4j.driver.Record, TargetEventPublication> incompleteMapping() {
return (typeSystem, driverRecord) -> mapRecordToPublication(typeSystem, driverRecord, EVENT_PUBLICATION_NODE);
}
var publicationNode = record.get(EVENT_PUBLICATION_NODE.getRequiredSymbolicName().getValue()).asNode();
private BiFunction<TypeSystem, org.neo4j.driver.Record, TargetEventPublication> 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);

20
spring-modulith-events/spring-modulith-events-neo4j/src/test/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationRepositoryTest.java
View File

@@ -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);
}
}
}