GH-301 - Add Neo4j event publication repository.
This commit is contained in:
Gerrit Meier
committed by
Oliver Drotbohm
Oliver Drotbohm
parent
bf404a571c
commit
4e069b98ea
@@ -23,6 +23,7 @@
|
||||
<module>spring-modulith-events-jpa</module>
|
||||
<module>spring-modulith-events-kafka</module>
|
||||
<module>spring-modulith-events-mongodb</module>
|
||||
<module>spring-modulith-events-neo4j</module>
|
||||
</modules>
|
||||
|
||||
<profiles>
|
||||
|
||||
58
spring-modulith-events/spring-modulith-events-neo4j/pom.xml
Normal file
58
spring-modulith-events/spring-modulith-events-neo4j/pom.xml
Normal file
@@ -0,0 +1,58 @@
|
||||
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
|
||||
<modelVersion>4.0.0</modelVersion>
|
||||
|
||||
<parent>
|
||||
<groupId>org.springframework.modulith</groupId>
|
||||
<artifactId>spring-modulith-events</artifactId>
|
||||
<version>1.1.0-SNAPSHOT</version>
|
||||
<relativePath>../pom.xml</relativePath>
|
||||
</parent>
|
||||
|
||||
<name>Spring Modulith - Events - Neo4j-based repository</name>
|
||||
<artifactId>spring-modulith-events-neo4j</artifactId>
|
||||
|
||||
<properties>
|
||||
<module.name>org.springframework.modulith.events.neo4j</module.name>
|
||||
</properties>
|
||||
|
||||
<dependencies>
|
||||
|
||||
<dependency>
|
||||
<groupId>${project.groupId}</groupId>
|
||||
<artifactId>spring-modulith-events-core</artifactId>
|
||||
<version>${project.version}</version>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.data</groupId>
|
||||
<artifactId>spring-data-neo4j</artifactId>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-test</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.springframework.boot</groupId>
|
||||
<artifactId>spring-boot-starter-data-neo4j</artifactId>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>neo4j</artifactId>
|
||||
<version>1.19.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
<dependency>
|
||||
<groupId>org.testcontainers</groupId>
|
||||
<artifactId>junit-jupiter</artifactId>
|
||||
<version>1.19.0</version>
|
||||
<scope>test</scope>
|
||||
</dependency>
|
||||
|
||||
</dependencies>
|
||||
</project>
|
||||
@@ -0,0 +1,30 @@
|
||||
package org.springframework.modulith.events.neo4j;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.util.UUID;
|
||||
|
||||
/**
|
||||
*
|
||||
* The event publication entity definition.
|
||||
*
|
||||
* @author Gerrit Meier
|
||||
*/
|
||||
public class Neo4jEventPublication {
|
||||
|
||||
public final UUID identifier;
|
||||
public final Instant publicationDate;
|
||||
public final String listenerId;
|
||||
public final Object event;
|
||||
public final String eventHash;
|
||||
|
||||
public Instant completionDate;
|
||||
|
||||
public Neo4jEventPublication(UUID identifier, Instant publicationDate, String listenerId, Object event, String eventHash) {
|
||||
this.identifier = identifier;
|
||||
this.publicationDate = publicationDate;
|
||||
this.listenerId = listenerId;
|
||||
this.event = event;
|
||||
this.eventHash = eventHash;
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,37 @@
|
||||
package org.springframework.modulith.events.neo4j;
|
||||
|
||||
import org.neo4j.cypherdsl.core.renderer.Configuration;
|
||||
import org.springframework.boot.autoconfigure.AutoConfiguration;
|
||||
import org.springframework.boot.autoconfigure.AutoConfigureBefore;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnMissingBean;
|
||||
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.data.neo4j.core.Neo4jClient;
|
||||
import org.springframework.modulith.events.config.EventPublicationAutoConfiguration;
|
||||
import org.springframework.modulith.events.config.EventPublicationConfigurationExtension;
|
||||
import org.springframework.modulith.events.core.EventSerializer;
|
||||
|
||||
/**
|
||||
* @author Gerrit Meier
|
||||
*/
|
||||
@AutoConfiguration
|
||||
@AutoConfigureBefore(EventPublicationAutoConfiguration.class)
|
||||
public class Neo4jEventPublicationAutoConfiguration implements EventPublicationConfigurationExtension {
|
||||
|
||||
@Bean
|
||||
Neo4jEventPublicationRepository neo4jEventPublicationRepository(Neo4jClient neo4jClient, Configuration cypherDslConfiguration, EventSerializer eventSerializer) {
|
||||
return new Neo4jEventPublicationRepository(neo4jClient, cypherDslConfiguration, eventSerializer);
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnMissingBean(Configuration.class)
|
||||
Configuration cypherDslConfiguration() {
|
||||
return Configuration.defaultConfig();
|
||||
}
|
||||
|
||||
@Bean
|
||||
@ConditionalOnProperty(name = "spring.modulith.events.neo4j.event-index.enabled", havingValue = "true")
|
||||
Neo4jIndexInitializer neo4jIndexInitializer(Neo4jClient neo4jClient) {
|
||||
return new Neo4jIndexInitializer(neo4jClient);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,310 @@
|
||||
package org.springframework.modulith.events.neo4j;
|
||||
|
||||
import org.neo4j.cypherdsl.core.Cypher;
|
||||
import org.neo4j.cypherdsl.core.Node;
|
||||
import org.neo4j.cypherdsl.core.Statement;
|
||||
import org.neo4j.cypherdsl.core.renderer.Configuration;
|
||||
import org.neo4j.cypherdsl.core.renderer.Renderer;
|
||||
import org.neo4j.driver.Values;
|
||||
import org.neo4j.driver.types.TypeSystem;
|
||||
import org.springframework.data.neo4j.core.Neo4jClient;
|
||||
import org.springframework.modulith.events.core.EventPublicationRepository;
|
||||
import org.springframework.modulith.events.core.EventSerializer;
|
||||
import org.springframework.modulith.events.core.PublicationTargetIdentifier;
|
||||
import org.springframework.modulith.events.core.TargetEventPublication;
|
||||
import org.springframework.transaction.annotation.Transactional;
|
||||
import org.springframework.util.Assert;
|
||||
import org.springframework.util.DigestUtils;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.time.ZoneOffset;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.Objects;
|
||||
import java.util.Optional;
|
||||
import java.util.UUID;
|
||||
|
||||
@Transactional
|
||||
class Neo4jEventPublicationRepository implements EventPublicationRepository {
|
||||
|
||||
private static final String ID = "identifier";
|
||||
private static final String EVENT_SERIALIZED = "eventSerialized";
|
||||
private static final String EVENT_HASH = "eventHash";
|
||||
private static final String EVENT_TYPE = "eventType";
|
||||
private static final String LISTENER_ID = "listenerId";
|
||||
private static final String PUBLICATION_DATE = "publicationDate";
|
||||
private static final String COMPLETION_DATE = "completionDate";
|
||||
|
||||
private static final Node eventPublicationNode = Cypher.node("Neo4jEventPublication").named("neo4jEventPublication");
|
||||
|
||||
private final Neo4jClient neo4jClient;
|
||||
private final Configuration cypherDslConfiguration;
|
||||
private final EventSerializer eventSerializer;
|
||||
|
||||
Neo4jEventPublicationRepository(Neo4jClient neo4jClient, Configuration cypherDslConfiguration, EventSerializer eventSerializer) {
|
||||
|
||||
Assert.notNull(neo4jClient, "Neo4jClient must not be null!");
|
||||
Assert.notNull(cypherDslConfiguration, "CypherDSL configuration must not be null!");
|
||||
Assert.notNull(eventSerializer, "EventSerializer must not be null!");
|
||||
|
||||
this.neo4jClient = neo4jClient;
|
||||
this.cypherDslConfiguration = cypherDslConfiguration;
|
||||
this.eventSerializer = eventSerializer;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.modulith.events.EventPublicationRepository#create(org.springframework.modulith.events.EventPublication)
|
||||
*/
|
||||
@Override
|
||||
@Transactional
|
||||
public TargetEventPublication create(TargetEventPublication publication) {
|
||||
|
||||
var identifier = publication.getIdentifier();
|
||||
var publicationDate = publication.getPublicationDate();
|
||||
var listenerId = publication.getTargetIdentifier().getValue();
|
||||
var event = publication.getEvent();
|
||||
var eventType = event.getClass().getName();
|
||||
|
||||
var eventSerialized = (String) eventSerializer.serialize(event);
|
||||
var eventHash = DigestUtils.md5DigestAsHex(eventSerialized.getBytes());
|
||||
|
||||
var createStatement = Cypher.create(eventPublicationNode)
|
||||
.set(eventPublicationNode.property(ID).to(Cypher.parameter(ID)))
|
||||
.set(eventPublicationNode.property(EVENT_SERIALIZED).to(Cypher.parameter(EVENT_SERIALIZED)))
|
||||
.set(eventPublicationNode.property(EVENT_HASH).to(Cypher.parameter(EVENT_HASH)))
|
||||
.set(eventPublicationNode.property(EVENT_TYPE).to(Cypher.parameter(EVENT_TYPE)))
|
||||
.set(eventPublicationNode.property(LISTENER_ID).to(Cypher.parameter(LISTENER_ID)))
|
||||
.set(eventPublicationNode.property(PUBLICATION_DATE).to(Cypher.parameter(PUBLICATION_DATE)))
|
||||
.build();
|
||||
|
||||
neo4jClient.query(renderStatement(createStatement))
|
||||
.bindAll(Map.of(
|
||||
ID, Values.value(identifier.toString()),
|
||||
EVENT_SERIALIZED, eventSerialized,
|
||||
EVENT_HASH, eventHash,
|
||||
EVENT_TYPE, eventType,
|
||||
LISTENER_ID, listenerId,
|
||||
PUBLICATION_DATE, Values.value(publicationDate.atOffset(ZoneOffset.UTC))
|
||||
))
|
||||
.run();
|
||||
|
||||
return publication;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.modulith.events.EventPublicationRepository#markCompleted(java.lang.Object, org.springframework.modulith.events.PublicationTargetIdentifier, java.time.Instant)
|
||||
*/
|
||||
@Override
|
||||
@Transactional
|
||||
public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) {
|
||||
var eventHash = DigestUtils.md5DigestAsHex(((String) eventSerializer.serialize(event)).getBytes());
|
||||
|
||||
var completeStatement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(EVENT_HASH).eq(Cypher.parameter(EVENT_HASH)))
|
||||
.and(eventPublicationNode.property(LISTENER_ID).eq(Cypher.parameter(LISTENER_ID)))
|
||||
.set(eventPublicationNode.property(COMPLETION_DATE).to(Cypher.parameter(COMPLETION_DATE)))
|
||||
.build();
|
||||
|
||||
neo4jClient.query(renderStatement(completeStatement))
|
||||
.bind(eventHash).to(EVENT_HASH)
|
||||
.bind(identifier.getValue()).to(LISTENER_ID)
|
||||
.bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE)
|
||||
.run();
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier)
|
||||
*/
|
||||
@Override
|
||||
@Transactional(readOnly = true)
|
||||
public List<TargetEventPublication> findIncompletePublications() {
|
||||
|
||||
var findIncompleteStatement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(COMPLETION_DATE).isNull())
|
||||
.returning(eventPublicationNode)
|
||||
.orderBy(eventPublicationNode.property(PUBLICATION_DATE))
|
||||
.build();
|
||||
|
||||
return List.copyOf(neo4jClient.query(renderStatement(findIncompleteStatement))
|
||||
.fetchAs(TargetEventPublication.class)
|
||||
.mappedBy(this::mapRecordToPublication)
|
||||
.all());
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional(readOnly = true)
|
||||
public List<TargetEventPublication> findIncompletePublicationsPublishedBefore(Instant instant) {
|
||||
var findIncompleteStatement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(PUBLICATION_DATE).lt(Cypher.parameter(PUBLICATION_DATE)))
|
||||
.and(eventPublicationNode.property(COMPLETION_DATE).isNull())
|
||||
.returning(eventPublicationNode)
|
||||
.orderBy(eventPublicationNode.property(PUBLICATION_DATE))
|
||||
.build();
|
||||
|
||||
return List.copyOf(neo4jClient.query(renderStatement(findIncompleteStatement))
|
||||
.bind(Values.value(instant.atOffset(ZoneOffset.UTC))).to(PUBLICATION_DATE)
|
||||
.fetchAs(TargetEventPublication.class)
|
||||
.mappedBy(this::mapRecordToPublication)
|
||||
.all());
|
||||
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional(readOnly = true)
|
||||
public Optional<TargetEventPublication> findIncompletePublicationsByEventAndTargetIdentifier(Object event, PublicationTargetIdentifier targetIdentifier) {
|
||||
|
||||
var eventHash = DigestUtils.md5DigestAsHex(((String) eventSerializer.serialize(event)).getBytes());
|
||||
var listenerId = targetIdentifier.getValue();
|
||||
|
||||
var statement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(EVENT_HASH).eq(Cypher.parameter(EVENT_HASH)))
|
||||
.and(eventPublicationNode.property(LISTENER_ID).eq(Cypher.parameter(LISTENER_ID)))
|
||||
.and(eventPublicationNode.property(COMPLETION_DATE).isNull())
|
||||
.returning(eventPublicationNode)
|
||||
.build();
|
||||
|
||||
return neo4jClient.query(renderStatement(statement))
|
||||
.bindAll(Map.of(EVENT_HASH, eventHash, LISTENER_ID, listenerId))
|
||||
.fetchAs(TargetEventPublication.class)
|
||||
.mappedBy(this::mapRecordToPublication)
|
||||
.one();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void deletePublications(List<UUID> identifiers) {
|
||||
var deleteStatement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(ID).in(Cypher.parameter(ID)))
|
||||
.delete(eventPublicationNode)
|
||||
.build();
|
||||
|
||||
neo4jClient.query(renderStatement(deleteStatement))
|
||||
.bind(identifiers.stream().map(UUID::toString).toList()).to(ID)
|
||||
.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void deleteCompletedPublications() {
|
||||
var deleteStatement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(COMPLETION_DATE).isNotNull())
|
||||
.delete(eventPublicationNode)
|
||||
.build();
|
||||
|
||||
neo4jClient.query(renderStatement(deleteStatement))
|
||||
.run();
|
||||
}
|
||||
|
||||
@Override
|
||||
@Transactional
|
||||
public void deleteCompletedPublicationsBefore(Instant instant) {
|
||||
var deleteStatement = Cypher.match(eventPublicationNode)
|
||||
.where(eventPublicationNode.property(PUBLICATION_DATE).lt(Cypher.parameter(PUBLICATION_DATE)))
|
||||
.and(eventPublicationNode.property(COMPLETION_DATE).isNotNull())
|
||||
.delete(eventPublicationNode)
|
||||
.build();
|
||||
|
||||
neo4jClient.query(renderStatement(deleteStatement))
|
||||
.bind(Values.value(instant.atOffset(ZoneOffset.UTC))).to(PUBLICATION_DATE)
|
||||
.run();
|
||||
}
|
||||
|
||||
private Neo4jEventPublicationAdapter mapRecordToPublication(TypeSystem typeSystem, org.neo4j.driver.Record record) {
|
||||
var publicationNode = record.get(eventPublicationNode.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();
|
||||
|
||||
try {
|
||||
Object event = eventSerializer.deserialize(eventSerialized, Class.forName(eventType));
|
||||
Neo4jEventPublication publication = new Neo4jEventPublication(identifier, publicationDate, listenerId, event, eventHash);
|
||||
|
||||
return new Neo4jEventPublicationAdapter(publication);
|
||||
} catch (ClassNotFoundException e) {
|
||||
throw new RuntimeException(e);
|
||||
}
|
||||
}
|
||||
|
||||
private String renderStatement(Statement statement) {
|
||||
return Renderer.getRenderer(cypherDslConfiguration).render(statement);
|
||||
}
|
||||
|
||||
|
||||
public static class Neo4jEventPublicationAdapter implements TargetEventPublication {
|
||||
|
||||
private final Neo4jEventPublication delegate;
|
||||
|
||||
public Neo4jEventPublicationAdapter(Neo4jEventPublication delegate) {
|
||||
this.delegate = delegate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public UUID getIdentifier() {
|
||||
return delegate.identifier;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Object getEvent() {
|
||||
return delegate.event;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Instant getPublicationDate() {
|
||||
return delegate.publicationDate;
|
||||
}
|
||||
|
||||
@Override
|
||||
public Optional<Instant> getCompletionDate() {
|
||||
return Optional.ofNullable(delegate.completionDate);
|
||||
}
|
||||
|
||||
@Override
|
||||
public void markCompleted(Instant instant) {
|
||||
delegate.completionDate = instant;
|
||||
}
|
||||
|
||||
@Override
|
||||
public PublicationTargetIdentifier getTargetIdentifier() {
|
||||
return PublicationTargetIdentifier.of(delegate.listenerId);
|
||||
}
|
||||
|
||||
@Override
|
||||
public boolean isPublicationCompleted() {
|
||||
return delegate.completionDate != null;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see java.lang.Object#equals(java.lang.Object)
|
||||
*/
|
||||
@Override
|
||||
public boolean equals(Object obj) {
|
||||
|
||||
if (this == obj) {
|
||||
return true;
|
||||
}
|
||||
|
||||
if (!(obj instanceof Neo4jEventPublicationAdapter that)) {
|
||||
return false;
|
||||
}
|
||||
|
||||
return Objects.equals(delegate, that.delegate);
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see java.lang.Object#hashCode()
|
||||
*/
|
||||
@Override
|
||||
public int hashCode() {
|
||||
return Objects.hash(delegate);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,23 @@
|
||||
package org.springframework.modulith.events.neo4j;
|
||||
|
||||
import org.springframework.beans.factory.InitializingBean;
|
||||
import org.springframework.data.neo4j.core.Neo4jClient;
|
||||
|
||||
/**
|
||||
* @author Gerrit Meier
|
||||
*/
|
||||
public class Neo4jIndexInitializer implements InitializingBean {
|
||||
|
||||
private final Neo4jClient neo4jClient;
|
||||
|
||||
public Neo4jIndexInitializer(Neo4jClient neo4jClient) {
|
||||
this.neo4jClient = neo4jClient;
|
||||
}
|
||||
|
||||
@Override
|
||||
public void afterPropertiesSet() throws Exception {
|
||||
neo4jClient
|
||||
.query("CREATE INDEX eventHashIndex IF NOT EXISTS FOR (n:`Neo4jEventPublication`) ON (n.eventHash)")
|
||||
.run();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
{
|
||||
"properties": [
|
||||
{
|
||||
"name": "spring.modulith.events.neo4j.event-index.enabled",
|
||||
"type": "java.lang.boolean",
|
||||
"description": "Whether to initialize the index on the Neo4j event publication event hash property.",
|
||||
"defaultValue": "false"
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1 @@
|
||||
org.springframework.modulith.events.neo4j.Neo4jEventPublicationAutoConfiguration
|
||||
@@ -0,0 +1,230 @@
|
||||
package org.springframework.modulith.events.neo4j;
|
||||
|
||||
import lombok.Value;
|
||||
import org.junit.jupiter.api.BeforeEach;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.neo4j.cypherdsl.core.renderer.Dialect;
|
||||
import org.neo4j.driver.AuthTokens;
|
||||
import org.neo4j.driver.Driver;
|
||||
import org.neo4j.driver.GraphDatabase;
|
||||
import org.neo4j.driver.types.Node;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.test.mock.mockito.MockBean;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.context.annotation.Import;
|
||||
import org.springframework.modulith.events.core.EventSerializer;
|
||||
import org.springframework.modulith.events.core.PublicationTargetIdentifier;
|
||||
import org.springframework.modulith.events.core.TargetEventPublication;
|
||||
import org.springframework.modulith.testapp.TestApplication;
|
||||
import org.springframework.test.context.junit.jupiter.SpringJUnitConfig;
|
||||
import org.springframework.util.DigestUtils;
|
||||
import org.testcontainers.containers.Neo4jContainer;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
import org.testcontainers.utility.DockerImageName;
|
||||
|
||||
import java.time.Instant;
|
||||
import java.time.temporal.ChronoUnit;
|
||||
import java.util.List;
|
||||
import java.util.UUID;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
import static org.mockito.Mockito.when;
|
||||
|
||||
/**
|
||||
* @author Gerrit Meier
|
||||
*/
|
||||
@SpringJUnitConfig(Neo4jEventPublicationRepositoryTest.Config.class)
|
||||
@Testcontainers(disabledWithoutDocker = true)
|
||||
class Neo4jEventPublicationRepositoryTest {
|
||||
|
||||
@Container
|
||||
private static Neo4jContainer<?> neo4jContainer = new Neo4jContainer<>(DockerImageName.parse("neo4j:5"))
|
||||
.withRandomPassword();
|
||||
|
||||
static final PublicationTargetIdentifier TARGET_IDENTIFIER = PublicationTargetIdentifier.of("listener");
|
||||
|
||||
@Autowired
|
||||
private Neo4jEventPublicationRepository repository;
|
||||
|
||||
@Autowired
|
||||
private Driver driver;
|
||||
|
||||
@MockBean
|
||||
private EventSerializer eventSerializer;
|
||||
|
||||
@BeforeEach
|
||||
void clearDb() {
|
||||
try (var session = driver.session()) {
|
||||
session.run("MATCH (n) detach delete n").consume();
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void createEventPublication() {
|
||||
var testEvent = new TestEvent("id");
|
||||
var eventSerialized = "{\"eventId\":\"id\"}";
|
||||
var eventHash = DigestUtils.md5DigestAsHex(eventSerialized.getBytes());
|
||||
|
||||
when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
|
||||
var publication = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));
|
||||
|
||||
try (var session = driver.session()) {
|
||||
var result = session.run("MATCH (p:Neo4jEventPublication) return p")
|
||||
.single();
|
||||
|
||||
Node neo4jEventPublicationNode = result.get("p").asNode();
|
||||
assertThat(UUID.fromString(neo4jEventPublicationNode.get("identifier").asString())).isEqualTo(publication.getIdentifier());
|
||||
assertThat(neo4jEventPublicationNode.get("publicationDate").asZonedDateTime().toInstant()).isEqualTo(publication.getPublicationDate());
|
||||
assertThat(neo4jEventPublicationNode.get("listenerId").asString()).isEqualTo(publication.getTargetIdentifier().getValue());
|
||||
assertThat(neo4jEventPublicationNode.get("completionDate").isNull()).isTrue();
|
||||
assertThat(neo4jEventPublicationNode.get("eventSerialized").asString()).isEqualTo(eventSerialized);
|
||||
assertThat(neo4jEventPublicationNode.get("eventHash").asString()).isEqualTo(eventHash);
|
||||
}
|
||||
|
||||
}
|
||||
|
||||
@Test
|
||||
void updateEventPublication() {
|
||||
var testEvent1 = new TestEvent("id1");
|
||||
var event1Serialized = "{\"eventId\":\"id1\"}";
|
||||
var testEvent2 = new TestEvent("id2");
|
||||
var event2Serialized = "{\"eventId\":\"id2\"}";
|
||||
|
||||
when(eventSerializer.serialize(testEvent1)).thenReturn(event1Serialized);
|
||||
when(eventSerializer.serialize(testEvent2)).thenReturn(event2Serialized);
|
||||
when(eventSerializer.deserialize(event2Serialized, TestEvent.class)).thenReturn(testEvent2);
|
||||
|
||||
var event1 = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER));
|
||||
var event2 = repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER));
|
||||
|
||||
var now = Instant.now();
|
||||
repository.markCompleted(event1, now);
|
||||
|
||||
assertThat(repository.findIncompletePublications()).hasSize(1)
|
||||
.element(0)
|
||||
.extracting(TargetEventPublication::getEvent).isEqualTo(event2.getEvent());
|
||||
}
|
||||
|
||||
@Test
|
||||
void findInCompletePastPublications() {
|
||||
var testEvent = new TestEvent("id");
|
||||
var eventSerialized = "{\"eventId\":\"id\"}";
|
||||
|
||||
when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
|
||||
when(eventSerializer.deserialize(eventSerialized, TestEvent.class)).thenReturn(testEvent);
|
||||
|
||||
var event = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));
|
||||
|
||||
var newer = Instant.now().plus(1L, ChronoUnit.MINUTES);
|
||||
var older = Instant.now().minus(1L, ChronoUnit.MINUTES);
|
||||
|
||||
assertThat(repository.findIncompletePublicationsPublishedBefore(newer)).hasSize(1)
|
||||
.element(0)
|
||||
.extracting(TargetEventPublication::getEvent).isEqualTo(event.getEvent());
|
||||
|
||||
assertThat(repository.findIncompletePublicationsPublishedBefore(older)).hasSize(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void findIncompleteByEventAndTargetIdentifier() {
|
||||
var testEvent = new TestEvent("id");
|
||||
var eventSerialized = "{\"eventId\":\"id\"}";
|
||||
|
||||
when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
|
||||
when(eventSerializer.deserialize(eventSerialized, TestEvent.class)).thenReturn(testEvent);
|
||||
|
||||
var event = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));
|
||||
|
||||
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, event.getTargetIdentifier()))
|
||||
.isPresent();
|
||||
}
|
||||
|
||||
@Test
|
||||
void deletePublicationById() {
|
||||
TestEvent testEvent = new TestEvent("id");
|
||||
var eventSerialized = "{\"eventId\":\"id\"}";
|
||||
when(eventSerializer.serialize(testEvent)).thenReturn(eventSerialized);
|
||||
|
||||
var event = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER));
|
||||
assertThat(repository.findIncompletePublications()).hasSize(1);
|
||||
|
||||
repository.deletePublications(List.of(event.getIdentifier()));
|
||||
assertThat(repository.findIncompletePublications()).hasSize(0);
|
||||
}
|
||||
|
||||
@Test
|
||||
void deleteCompletedPublications() {
|
||||
TestEvent testEvent1 = new TestEvent("id1");
|
||||
var event1Serialized = "{\"eventId\":\"id1\"}";
|
||||
TestEvent testEvent2 = new TestEvent("id2");
|
||||
var event2Serialized = "{\"eventId\":\"id2\"}";
|
||||
|
||||
when(eventSerializer.serialize(testEvent1)).thenReturn(event1Serialized);
|
||||
when(eventSerializer.serialize(testEvent2)).thenReturn(event2Serialized);
|
||||
when(eventSerializer.deserialize(event1Serialized, TestEvent.class)).thenReturn(testEvent1);
|
||||
when(eventSerializer.deserialize(event2Serialized, TestEvent.class)).thenReturn(testEvent2);
|
||||
|
||||
var event1 = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER));
|
||||
|
||||
repository.markCompleted(event1, Instant.now());
|
||||
|
||||
repository.deleteCompletedPublications();
|
||||
|
||||
try (var session = driver.session()) {
|
||||
var count = session.run("MATCH (n) WHERE n.completionDate is not null return count(n)").single().get("count(n)").asLong();
|
||||
assertThat(count).isEqualTo(0);
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
void deleteCompletedPublicationsBefore() throws Exception {
|
||||
TestEvent testEvent1 = new TestEvent("id1");
|
||||
var event1Serialized = "{\"eventId\":\"id1\"}";
|
||||
TestEvent testEvent2 = new TestEvent("id2");
|
||||
var event2Serialized = "{\"eventId\":\"id2\"}";
|
||||
|
||||
when(eventSerializer.serialize(testEvent1)).thenReturn(event1Serialized);
|
||||
when(eventSerializer.serialize(testEvent2)).thenReturn(event2Serialized);
|
||||
when(eventSerializer.deserialize(event1Serialized, TestEvent.class)).thenReturn(testEvent1);
|
||||
when(eventSerializer.deserialize(event2Serialized, TestEvent.class)).thenReturn(testEvent2);
|
||||
|
||||
var event1 = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER));
|
||||
|
||||
Instant old = Instant.now();
|
||||
repository.markCompleted(event1, old);
|
||||
Thread.sleep(100);
|
||||
var event2 = repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER));
|
||||
repository.markCompleted(event2, Instant.now());
|
||||
|
||||
repository.deleteCompletedPublicationsBefore(old.plus(10, ChronoUnit.MILLIS));
|
||||
// defensive check just to be sure
|
||||
assertThat(repository.findIncompletePublications()).hasSize(0);
|
||||
|
||||
try (var session = driver.session()) {
|
||||
var records = session.run("MATCH (n) WHERE n.completionDate is not null return n").list();
|
||||
assertThat(records.size()).isEqualTo(1);
|
||||
assertThat(records.get(0).get("n").asNode().get("eventSerialized").asString()).contains("id2");
|
||||
}
|
||||
}
|
||||
|
||||
@Value
|
||||
private static final class TestEvent {
|
||||
String eventId;
|
||||
}
|
||||
|
||||
@Import({TestApplication.class})
|
||||
@Configuration
|
||||
static class Config {
|
||||
@Bean
|
||||
public Driver driver() {
|
||||
return GraphDatabase.driver(neo4jContainer.getBoltUrl(), AuthTokens.basic("neo4j", neo4jContainer.getAdminPassword()));
|
||||
}
|
||||
|
||||
@Bean
|
||||
public org.neo4j.cypherdsl.core.renderer.Configuration cypherDslConfiguration() {
|
||||
return org.neo4j.cypherdsl.core.renderer.Configuration.newConfig().withDialect(Dialect.NEO4J_5).build();
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,99 @@
|
||||
package org.springframework.modulith.events.neo4j;
|
||||
|
||||
import org.junit.jupiter.api.Nested;
|
||||
import org.junit.jupiter.api.Test;
|
||||
import org.neo4j.driver.AuthTokens;
|
||||
import org.neo4j.driver.Driver;
|
||||
import org.neo4j.driver.GraphDatabase;
|
||||
import org.springframework.beans.factory.annotation.Autowired;
|
||||
import org.springframework.boot.autoconfigure.ImportAutoConfiguration;
|
||||
import org.springframework.boot.test.autoconfigure.data.neo4j.DataNeo4jTest;
|
||||
import org.springframework.boot.test.mock.mockito.MockBean;
|
||||
import org.springframework.context.annotation.Bean;
|
||||
import org.springframework.context.annotation.Configuration;
|
||||
import org.springframework.data.neo4j.core.Neo4jClient;
|
||||
import org.springframework.modulith.events.core.EventSerializer;
|
||||
import org.springframework.modulith.testapp.TestApplication;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.testcontainers.containers.Neo4jContainer;
|
||||
import org.testcontainers.junit.jupiter.Container;
|
||||
import org.testcontainers.junit.jupiter.Testcontainers;
|
||||
import org.testcontainers.utility.DockerImageName;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
import static org.assertj.core.api.Assertions.assertThat;
|
||||
|
||||
/**
|
||||
* @author Gerrit Meier
|
||||
*/
|
||||
public class Neo4jIndexInitializerTest {
|
||||
|
||||
@ImportAutoConfiguration({Neo4jEventPublicationAutoConfiguration.class, TestBase.Config.class})
|
||||
@Testcontainers(disabledWithoutDocker = true)
|
||||
@ContextConfiguration(classes = TestApplication.class)
|
||||
static class TestBase {
|
||||
|
||||
@Container
|
||||
private static Neo4jContainer<?> neo4jContainer = new Neo4jContainer<>(DockerImageName.parse("neo4j:5"))
|
||||
.withRandomPassword();
|
||||
|
||||
@MockBean
|
||||
EventSerializer eventSerializer;
|
||||
|
||||
|
||||
@Configuration
|
||||
static class Config {
|
||||
|
||||
@Bean
|
||||
Driver driver() {
|
||||
return GraphDatabase.driver(neo4jContainer.getBoltUrl(), AuthTokens.basic("neo4j", neo4jContainer.getAdminPassword()));
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@DataNeo4jTest(properties = "spring.modulith.events.neo4j.event-index.enabled=true")
|
||||
class WithIndexEnabled extends TestBase {
|
||||
@Autowired
|
||||
Neo4jClient neo4jClient;
|
||||
|
||||
@Autowired
|
||||
Optional<Neo4jIndexInitializer> neo4jIndexInitializer;
|
||||
|
||||
@Test
|
||||
void indexInitializerBeanIsPresent() {
|
||||
assertThat(neo4jIndexInitializer).isPresent();
|
||||
}
|
||||
|
||||
@Test
|
||||
void indexWasCreated() {
|
||||
assertThat(neo4jClient.query("SHOW INDEX YIELD name")
|
||||
.fetchAs(String.class)
|
||||
.all()).contains("eventHashIndex");
|
||||
}
|
||||
}
|
||||
|
||||
@Nested
|
||||
@DataNeo4jTest(properties = "spring.modulith.events.neo4j.event-index.enabled=false")
|
||||
class WithoutIndexEnabled extends TestBase {
|
||||
@Autowired
|
||||
Neo4jClient neo4jClient;
|
||||
|
||||
@Autowired
|
||||
Optional<Neo4jIndexInitializer> neo4jIndexInitializer;
|
||||
|
||||
@Test
|
||||
void indexInitializerBeanIsNotPresent() {
|
||||
assertThat(neo4jIndexInitializer).isEmpty();
|
||||
}
|
||||
|
||||
@Test
|
||||
void indexWasNotCreated() {
|
||||
assertThat(neo4jClient.query("SHOW INDEX YIELD name")
|
||||
.fetchAs(String.class)
|
||||
.all()).doesNotContain("eventHashIndex");
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -0,0 +1,10 @@
|
||||
package org.springframework.modulith.testapp;
|
||||
|
||||
import org.springframework.boot.autoconfigure.SpringBootApplication;
|
||||
|
||||
/**
|
||||
* @author Gerrit Meier
|
||||
*/
|
||||
@SpringBootApplication
|
||||
public class TestApplication {
|
||||
}
|
||||
Reference in New Issue
Block a user