From cd0b5cf5c37ba6237d733c992a2c12a1679bbf5e Mon Sep 17 00:00:00 2001 From: Oliver Drotbohm Date: Tue, 27 Aug 2024 21:52:53 +0200 Subject: [PATCH] GH-748 - Allow publication completion to delete database entries. We now expose a spring.modulith.events.completion-mode property, defaulting the previous behavior to a value of UPDATE. The property can also be configured to DELETE, which will cause the persistence implementations to flip to removing the database entries for event publications instead of setting the completion date. --- .../events/support/CompletionMode.java | 57 ++++++++ .../spring-configuration-metadata.json | 6 + ...JdbcEventPublicationAutoConfiguration.java | 14 +- .../jdbc/JdbcEventPublicationRepository.java | 64 ++++++--- .../events/jdbc/JdbcRepositorySettings.java | 77 +++++++++++ ...PublicationRepositoryIntegrationTests.java | 130 ++++++++++++------ .../jpa/JpaEventPublicationConfiguration.java | 7 +- .../jpa/JpaEventPublicationRepository.java | 64 +++++++-- ...PublicationRepositoryIntegrationTests.java | 60 +++++--- ...goDbEventPublicationAutoConfiguration.java | 6 +- .../MongoDbEventPublicationRepository.java | 33 ++++- ...MongoDbEventPublicationRepositoryTest.java | 47 +++++-- ...eo4jEventPublicationAutoConfiguration.java | 7 +- .../Neo4jEventPublicationRepository.java | 47 +++++-- .../Neo4jEventPublicationRepositoryTest.java | 64 +++++++-- .../antora/modules/ROOT/pages/appendix.adoc | 4 + .../antora/modules/ROOT/pages/events.adoc | 14 ++ 17 files changed, 572 insertions(+), 129 deletions(-) create mode 100644 spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionMode.java create mode 100644 spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcRepositorySettings.java diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionMode.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionMode.java new file mode 100644 index 00000000..37158e8e --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionMode.java @@ -0,0 +1,57 @@ +/* + * Copyright 2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.modulith.events.support; + +import org.springframework.core.env.Environment; +import org.springframework.util.Assert; + +/** + * Different modes of event completion. + * + * @author Oliver Drotbohm + * @since 1.3 + * @soundtrack Lettuce - Waffles (Unify) + */ +public enum CompletionMode { + + /** + * Completes an {@link org.springframework.modulith.events.EventPublication} by setting its completion date and + * updating the database entry accordingly. + */ + UPDATE, + + /** + * Completes an {@link org.springframework.modulith.events.EventPublication} by removing the database entry. + */ + DELETE; + + public static final String PROPERTY = "spring.modulith.events.completion-mode"; + + /** + * Looks up the {@link CompletionMode} from the given environment or uses {@link #UPDATE} as default. + * + * @param environment must not be {@literal null}. + * @return will never be {@literal null}. + */ + public static CompletionMode from(Environment environment) { + + Assert.notNull(environment, "Environment must not be null!"); + + var result = environment.getProperty(PROPERTY, CompletionMode.class); + + return result == null ? CompletionMode.UPDATE : result; + } +} diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/resources/META-INF/spring-configuration-metadata.json b/spring-modulith-events/spring-modulith-events-core/src/main/resources/META-INF/spring-configuration-metadata.json index 28691d94..6915b97d 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/resources/META-INF/spring-configuration-metadata.json +++ b/spring-modulith-events/spring-modulith-events-core/src/main/resources/META-INF/spring-configuration-metadata.json @@ -17,6 +17,12 @@ "type": "java.lang.Boolean", "description": "Whether to enable event externalization.", "defaultValue": "true" + }, + { + "name": "spring.modulith.events.completion-mode", + "type": "org.springframework.modulith.events.support.CompletionMode", + "description": "How to complete event publications.", + "defaultValue": "update" } ] } diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationAutoConfiguration.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationAutoConfiguration.java index 9af09d08..526be383 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationAutoConfiguration.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationAutoConfiguration.java @@ -24,12 +24,14 @@ import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.env.Environment; import org.springframework.core.io.ResourceLoader; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.jdbc.support.JdbcUtils; import org.springframework.modulith.events.config.EventPublicationAutoConfiguration; import org.springframework.modulith.events.config.EventPublicationConfigurationExtension; import org.springframework.modulith.events.core.EventSerializer; +import org.springframework.modulith.events.support.CompletionMode; /** * @author Dmitry Belyaev @@ -48,10 +50,16 @@ class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigura } @Bean - JdbcEventPublicationRepository jdbcEventPublicationRepository(JdbcTemplate jdbcTemplate, - EventSerializer serializer, DatabaseType databaseType, JdbcConfigurationProperties properties) { + JdbcRepositorySettings jdbcEventPublicationRepositorySettings(DatabaseType databaseType, + JdbcConfigurationProperties properties, Environment environment) { - return new JdbcEventPublicationRepository(jdbcTemplate, serializer, databaseType, properties); + return new JdbcRepositorySettings(databaseType, CompletionMode.from(environment), properties.getSchema()); + } + + @Bean + JdbcEventPublicationRepository jdbcEventPublicationRepository(JdbcTemplate jdbcTemplate, + EventSerializer serializer, JdbcRepositorySettings settings) { + return new JdbcEventPublicationRepository(jdbcTemplate, serializer, settings); } @Bean diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java index 7d127446..25cfa455 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java @@ -116,6 +116,20 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean ID IN """; + private static final String SQL_STATEMENT_DELETE_BY_EVENT_AND_LISTENER_ID = """ + DELETE FROM %s + WHERE + LISTENER_ID = ? + AND SERIALIZED_EVENT = ? + """; + + private static final String SQL_STATEMENT_DELETE_BY_ID = """ + DELETE + FROM %s + WHERE + ID = ? + """; + private static final String SQL_STATEMENT_DELETE_UNCOMPLETED = """ DELETE FROM %s @@ -134,7 +148,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean private final JdbcOperations operations; private final EventSerializer serializer; - private final DatabaseType databaseType; + private final JdbcRepositorySettings settings; + private ClassLoader classLoader; private final String sqlStatementInsert, @@ -145,6 +160,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean sqlStatementUpdateById, sqlStatementFindByEventAndListenerId, sqlStatementDelete, + sqlStatementDeleteByEventAndListenerId, + sqlStatementDeleteById, sqlStatementDeleteUncompleted, sqlStatementDeleteUncompletedBefore; @@ -154,22 +171,20 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean * * @param operations must not be {@literal null}. * @param serializer must not be {@literal null}. - * @param databaseType must not be {@literal null}. - * @param properties must not be {@literal null}. + * @param settings must not be {@literal null}. */ public JdbcEventPublicationRepository(JdbcOperations operations, EventSerializer serializer, - DatabaseType databaseType, JdbcConfigurationProperties properties) { + JdbcRepositorySettings settings) { Assert.notNull(operations, "JdbcOperations must not be null!"); Assert.notNull(serializer, "EventSerializer must not be null!"); - Assert.notNull(databaseType, "DatabaseType must not be null!"); - Assert.notNull(properties, "JdbcConfigurationProperties must not be null!"); + Assert.notNull(settings, "DatabaseType must not be null!"); this.operations = operations; this.serializer = serializer; - this.databaseType = databaseType; + this.settings = settings; - var schema = properties.getSchema(); + var schema = settings.getSchema(); var table = ObjectUtils.isEmpty(schema) ? "EVENT_PUBLICATION" : schema + ".EVENT_PUBLICATION"; this.sqlStatementInsert = SQL_STATEMENT_INSERT.formatted(table); @@ -180,6 +195,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean this.sqlStatementUpdateById = SQL_STATEMENT_UPDATE_BY_ID.formatted(table); this.sqlStatementFindByEventAndListenerId = SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID.formatted(table); this.sqlStatementDelete = SQL_STATEMENT_DELETE.formatted(table); + this.sqlStatementDeleteByEventAndListenerId = SQL_STATEMENT_DELETE_BY_EVENT_AND_LISTENER_ID.formatted(table); + this.sqlStatementDeleteById = SQL_STATEMENT_DELETE_BY_ID.formatted(table); this.sqlStatementDeleteUncompleted = SQL_STATEMENT_DELETE_UNCOMPLETED.formatted(table); this.sqlStatementDeleteUncompletedBefore = SQL_STATEMENT_DELETE_UNCOMPLETED_BEFORE.formatted(table); } @@ -222,10 +239,20 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean @Transactional public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) { - operations.update(sqlStatementUpdateByEventAndListenerId, // - Timestamp.from(completionDate), // - identifier.getValue(), // - serializer.serialize(event)); + var targetIdentifier = identifier.getValue(); + var serializedEvent = serializer.serialize(event); + + if (settings.isDeleteCompletion()) { + + operations.update(sqlStatementDeleteByEventAndListenerId, targetIdentifier, serializedEvent); + + } else { + + operations.update(sqlStatementUpdateByEventAndListenerId, // + Timestamp.from(completionDate), // + targetIdentifier, // + serializedEvent); + } } /* @@ -235,7 +262,12 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean @Override @Transactional public void markCompleted(UUID identifier, Instant completionDate) { - operations.update(sqlStatementUpdateById, Timestamp.from(completionDate), uuidToDatabase(identifier)); + + if (settings.isDeleteCompletion()) { + operations.update(sqlStatementDeleteById, uuidToDatabase(identifier)); + } else { + operations.update(sqlStatementUpdateById, Timestamp.from(completionDate), uuidToDatabase(identifier)); + } } /* @@ -294,7 +326,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean @Override public void deletePublications(List identifiers) { - var dbIdentifiers = identifiers.stream().map(databaseType::uuidToDatabase).toList(); + var dbIdentifiers = identifiers.stream().map(this::uuidToDatabase).toList(); batch(dbIdentifiers, DELETE_BATCH_SIZE) .forEach(it -> operations.update(sqlStatementDelete.concat(toParameterPlaceholders(it.length)), it)); @@ -376,11 +408,11 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean } private Object uuidToDatabase(UUID id) { - return databaseType.uuidToDatabase(id); + return settings.getDatabaseType().uuidToDatabase(id); } private UUID getUuidFromResultSet(ResultSet rs) throws SQLException { - return databaseType.databaseToUUID(rs.getObject("ID")); + return settings.getDatabaseType().databaseToUUID(rs.getObject("ID")); } @Nullable diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcRepositorySettings.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcRepositorySettings.java new file mode 100644 index 00000000..a297a954 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcRepositorySettings.java @@ -0,0 +1,77 @@ +/* + * Copyright 2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ +package org.springframework.modulith.events.jdbc; + +import org.springframework.lang.Nullable; +import org.springframework.modulith.events.support.CompletionMode; +import org.springframework.util.Assert; + +/** + * Internal abstraction of customization options for {@link JdbcEventPublicationRepository}. + * + * @author Oliver Drotbohm + * @since 1.3 + * @soundtrack Jeff Coffin - Bom Bom (Only the Horizon) + */ +public class JdbcRepositorySettings { + + private final DatabaseType databaseType; + private final String schema; + private final CompletionMode completionMode; + + /** + * Creates a new {@link JdbcRepositorySettings} for the given {@link DatabaseType}, {@link CompletionMode} and schema + * + * @param databaseType must not be {@literal null}. + * @param schema can be {@literal null} + * @param completionMode must not be {@literal null}. + */ + JdbcRepositorySettings(DatabaseType databaseType, CompletionMode completionMode, @Nullable String schema) { + + Assert.notNull(databaseType, "Database type must not be null!"); + Assert.notNull(completionMode, "Completion mode must not be null!"); + + this.databaseType = databaseType; + this.schema = schema; + this.completionMode = completionMode; + } + + /** + * Returns the {@link DatabaseType}. + * + * @return will never be {@literal null}. + */ + public DatabaseType getDatabaseType() { + return databaseType; + } + + /** + * Return the schema to be used. + * + * @return can be {@literal null}. + */ + @Nullable + public String getSchema() { + return schema; + } + + /** + * Returns whether we use the deleting completion mode. + */ + public boolean isDeleteCompletion() { + return completionMode == CompletionMode.DELETE; + } +} diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java index 27f043a9..55713082 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java @@ -16,11 +16,14 @@ package org.springframework.modulith.events.jdbc; import static org.assertj.core.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.*; import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; import lombok.Value; +import java.lang.annotation.Retention; +import java.lang.annotation.RetentionPolicy; import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; @@ -39,6 +42,7 @@ import org.springframework.jdbc.core.JdbcOperations; 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.events.support.CompletionMode; import org.springframework.modulith.testapp.TestApplication; import org.springframework.test.context.ActiveProfiles; import org.springframework.test.context.ContextConfiguration; @@ -63,6 +67,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { @Autowired JdbcOperations operations; @Autowired JdbcEventPublicationRepository repository; + @Autowired JdbcRepositorySettings properties; @MockBean EventSerializer serializer; @@ -237,6 +242,8 @@ class JdbcEventPublicationRepositoryIntegrationTests { @Test // GH-251 void shouldDeleteCompletedEventsBefore() { + assumeFalse(properties.isDeleteCompletion()); + var testEvent1 = new TestEvent("abc"); var serializedEvent1 = "{\"eventId\":\"abc\"}"; var testEvent2 = new TestEvent("def"); @@ -303,11 +310,19 @@ class JdbcEventPublicationRepositoryIntegrationTests { repository.markCompleted(publication, Instant.now()); - assertThat(repository.findCompletedPublications()) - .hasSize(1) - .element(0) - .extracting(TargetEventPublication::getEvent) - .isEqualTo(event); + if (properties.isDeleteCompletion()) { + + assertThat(repository.findCompletedPublications()).isEmpty(); + assertThat(repository.findIncompletePublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } } @Test // GH-258 @@ -318,9 +333,17 @@ class JdbcEventPublicationRepositoryIntegrationTests { repository.markCompleted(publication.getIdentifier(), Instant.now()); - assertThat(repository.findCompletedPublications()) - .extracting(TargetEventPublication::getIdentifier) - .containsExactly(publication.getIdentifier()); + if (properties.isDeleteCompletion()) { + + assertThat(repository.findIncompletePublications()).isEmpty(); + assertThat(repository.findCompletedPublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(publication.getIdentifier()); + } } @Test // GH-753 @@ -342,7 +365,9 @@ class JdbcEventPublicationRepositoryIntegrationTests { assertThat(publication.getEvent()).isSameAs(publication.getEvent()); } - abstract String table(); + String table() { + return "EVENT_PUBLICATION"; + } private TargetEventPublication createPublication(Object event) { @@ -357,13 +382,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { @Nested @JdbcTest(properties = "spring.modulith.events.jdbc.schema-initialization.enabled=true") - static class WithNoDefinedSchemaName extends TestBase { - - @Override - String table() { - return "EVENT_PUBLICATION"; - } - } + static class WithNoDefinedSchemaName extends TestBase {} @Nested @JdbcTest(properties = { "spring.modulith.events.jdbc.schema-initialization.enabled=true", @@ -387,60 +406,87 @@ class JdbcEventPublicationRepositoryIntegrationTests { } } - // HSQL @Nested - @ActiveProfiles("hsqldb") - @Testcontainers(disabledWithoutDocker = false) + @JdbcTest(properties = { "spring.modulith.events.jdbc.schema-initialization.enabled=true", + CompletionMode.PROPERTY + "=DELETE" }) + static class WithDeleteCompletion extends TestBase {} + + // HSQL + + @WithHsql class HsqlWithNoDefinedSchemaName extends WithNoDefinedSchemaName {} - @Nested - @ActiveProfiles("hsqldb") - @Testcontainers(disabledWithoutDocker = false) + @WithHsql class HsqlWithDefinedSchemaName extends WithDefinedSchemaName {} - @Nested - @ActiveProfiles("hsqldb") - @Testcontainers(disabledWithoutDocker = false) + @WithHsql class HsqlWithEmptySchemaName extends WithEmptySchemaName {} + @WithHsql + class HsqlWithEmptyDeleteCompletion extends WithDeleteCompletion {} + // H2 - @Nested - @ActiveProfiles("h2") - @Testcontainers(disabledWithoutDocker = false) + + @WithH2 class H2WithNoDefinedSchemaName extends WithNoDefinedSchemaName {} - @Nested - @ActiveProfiles("h2") - @Testcontainers(disabledWithoutDocker = false) + @WithH2 class H2WithDefinedSchemaName extends WithDefinedSchemaName {} - @Nested - @ActiveProfiles("h2") - @Testcontainers(disabledWithoutDocker = false) + @WithH2 class H2WithEmptySchemaName extends WithEmptySchemaName {} + @WithH2 + class H2WithEmptyDeleteCompletion extends WithDeleteCompletion {} + // Postgres - @Nested - @ActiveProfiles("postgres") + + @WithPostgres class PostgresWithNoDefinedSchemaName extends WithNoDefinedSchemaName {} - @Nested - @ActiveProfiles("postgres") + @WithPostgres class PostgresWithDefinedSchemaName extends WithDefinedSchemaName {} - @Nested - @ActiveProfiles("postgres") + @WithPostgres class PostgresWithEmptySchemaName extends WithEmptySchemaName {} + @WithPostgres + class PostgresWithDeleteCompletion extends WithDeleteCompletion {} + // MySQL - @Nested - @ActiveProfiles("mysql") + + @WithMySql class MysqlWithNoDefinedSchemaName extends WithNoDefinedSchemaName {} + @WithMySql + class MysqlWithDeleteCompletion extends WithDeleteCompletion {} + @Value private static final class TestEvent { String eventId; } private static final class Sample {} + + @Nested + @ActiveProfiles("h2") + @Testcontainers(disabledWithoutDocker = false) + @Retention(RetentionPolicy.RUNTIME) + @interface WithH2 {} + + @Nested + @ActiveProfiles("hsql") + @Testcontainers(disabledWithoutDocker = false) + @Retention(RetentionPolicy.RUNTIME) + @interface WithHsql {} + + @Nested + @ActiveProfiles("mysql") + @Retention(RetentionPolicy.RUNTIME) + @interface WithMySql {} + + @Nested + @ActiveProfiles("postgres") + @Retention(RetentionPolicy.RUNTIME) + @interface WithPostgres {} } diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationConfiguration.java b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationConfiguration.java index bded2d95..5c4ea64a 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationConfiguration.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationConfiguration.java @@ -19,8 +19,10 @@ import jakarta.persistence.EntityManager; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.core.env.Environment; import org.springframework.modulith.events.config.EventPublicationConfigurationExtension; import org.springframework.modulith.events.core.EventSerializer; +import org.springframework.modulith.events.support.CompletionMode; /** * @author Oliver Drotbohm @@ -31,7 +33,8 @@ import org.springframework.modulith.events.core.EventSerializer; class JpaEventPublicationConfiguration implements EventPublicationConfigurationExtension { @Bean - JpaEventPublicationRepository jpaEventPublicationRepository(EntityManager em, EventSerializer serializer) { - return new JpaEventPublicationRepository(em, serializer); + JpaEventPublicationRepository jpaEventPublicationRepository(EntityManager em, EventSerializer serializer, + Environment environment) { + return new JpaEventPublicationRepository(em, serializer, CompletionMode.from(environment)); } } diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java index 5558c0d2..2bb762f6 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java @@ -28,6 +28,7 @@ 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.modulith.events.support.CompletionMode; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; @@ -94,9 +95,20 @@ class JpaEventPublicationRepository implements EventPublicationRepository { private static final String DELETE = """ delete - from JpaEventPublication p - where - p.id in ?1 + from JpaEventPublication p + where p.id in ?1 + """; + + private static final String DELETE_BY_EVENT_AND_LISTENER_ID = """ + delete JpaEventPublication p + where p.serializedEvent = ?1 + and p.listenerId = ?2 + """; + + private static final String DELETE_BY_ID = """ + delete + from JpaEventPublication p + where p.id = ?1 """; private static final String DELETE_COMPLETED = """ @@ -117,6 +129,7 @@ class JpaEventPublicationRepository implements EventPublicationRepository { private final EntityManager entityManager; private final EventSerializer serializer; + private final CompletionMode completionMode; /** * Creates a new {@link JpaEventPublicationRepository} for the given {@link EntityManager} and @@ -125,13 +138,16 @@ class JpaEventPublicationRepository implements EventPublicationRepository { * @param entityManager must not be {@literal null}. * @param serializer must not be {@literal null}. */ - public JpaEventPublicationRepository(EntityManager entityManager, EventSerializer serializer) { + public JpaEventPublicationRepository(EntityManager entityManager, EventSerializer serializer, + CompletionMode completionMode) { Assert.notNull(entityManager, "EntityManager must not be null!"); Assert.notNull(serializer, "EventSerializer must not be null!"); + Assert.notNull(completionMode, "Completion mode must not be null!"); this.entityManager = entityManager; this.serializer = serializer; + this.completionMode = completionMode; } /* @@ -153,11 +169,24 @@ class JpaEventPublicationRepository implements EventPublicationRepository { @Override public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) { - entityManager.createQuery(MARK_COMPLETED_BY_EVENT_AND_LISTENER_ID) - .setParameter(1, serializeEvent(event)) - .setParameter(2, identifier.getValue()) - .setParameter(3, completionDate) - .executeUpdate(); + var serializedEvent = serializeEvent(event); + var identifierValue = identifier.getValue(); + + if (completionMode == CompletionMode.DELETE) { + + entityManager.createQuery(DELETE_BY_EVENT_AND_LISTENER_ID) + .setParameter(1, serializedEvent) + .setParameter(2, identifierValue) + .executeUpdate(); + + } else { + + entityManager.createQuery(MARK_COMPLETED_BY_EVENT_AND_LISTENER_ID) + .setParameter(1, serializedEvent) + .setParameter(2, identifierValue) + .setParameter(3, completionDate) + .executeUpdate(); + } } /* @@ -167,10 +196,19 @@ class JpaEventPublicationRepository implements EventPublicationRepository { @Override public void markCompleted(UUID identifier, Instant completionDate) { - entityManager.createQuery(MARK_COMPLETED_BY_ID) - .setParameter(1, identifier) - .setParameter(2, completionDate) - .executeUpdate(); + if (completionMode == CompletionMode.DELETE) { + + entityManager.createQuery(DELETE_BY_ID) + .setParameter(1, identifier) + .executeUpdate(); + + } else { + + entityManager.createQuery(MARK_COMPLETED_BY_ID) + .setParameter(1, identifier) + .setParameter(2, completionDate) + .executeUpdate(); + } } /* diff --git a/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java b/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java index 460e474a..cf4eebac 100644 --- a/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java @@ -16,12 +16,11 @@ package org.springframework.modulith.events.jpa; import static org.assertj.core.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.*; import static org.mockito.Mockito.*; import jakarta.persistence.EntityManager; import jakarta.persistence.EntityManagerFactory; -import lombok.RequiredArgsConstructor; -import lombok.Value; import java.time.Instant; import java.time.LocalDateTime; @@ -33,25 +32,29 @@ import java.util.UUID; import javax.sql.DataSource; import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; -import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.boot.test.context.SpringBootTest; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; +import org.springframework.core.env.Environment; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; 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.events.support.CompletionMode; import org.springframework.orm.jpa.JpaTransactionManager; import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean; import org.springframework.orm.jpa.SharedEntityManagerCreator; import org.springframework.orm.jpa.vendor.AbstractJpaVendorAdapter; import org.springframework.orm.jpa.vendor.HibernateJpaVendorAdapter; -import org.springframework.test.context.TestConstructor; -import org.springframework.test.context.TestConstructor.AutowireMode; -import org.springframework.test.context.junit.jupiter.SpringExtension; +import org.springframework.test.context.ContextConfiguration; +import org.springframework.test.context.TestPropertySource; import org.springframework.transaction.annotation.Transactional; /** @@ -59,10 +62,8 @@ import org.springframework.transaction.annotation.Transactional; * @author Dmitry Belyaev * @author Björn Kieling */ -@ExtendWith(SpringExtension.class) -@TestConstructor(autowireMode = AutowireMode.ALL) +@SpringBootTest @Transactional -@RequiredArgsConstructor class JpaEventPublicationRepositoryIntegrationTests { private static final PublicationTargetIdentifier TARGET_IDENTIFIER = PublicationTargetIdentifier.of("listener"); @@ -111,8 +112,16 @@ class JpaEventPublicationRepositoryIntegrationTests { } } - private final JpaEventPublicationRepository repository; - private final EntityManager em; + @Autowired JpaEventPublicationRepository repository; + @Autowired EntityManager em; + @Autowired Environment environment; + + CompletionMode completionMode; + + @BeforeEach + void init() { + this.completionMode = environment.getProperty(CompletionMode.PROPERTY, CompletionMode.class); + } @AfterEach public void flush() { @@ -212,6 +221,8 @@ class JpaEventPublicationRepositoryIntegrationTests { @Test // GH-251 void shouldDeleteCompletedEventsBefore() { + assumeTrue(completionMode == CompletionMode.UPDATE); + var testEvent1 = new TestEvent("abc"); var serializedEvent1 = "{\"eventId\":\"abc\"}"; var testEvent2 = new TestEvent("def"); @@ -278,11 +289,26 @@ class JpaEventPublicationRepositoryIntegrationTests { repository.markCompleted(publication, Instant.now()); - assertThat(repository.findCompletedPublications()) - .hasSize(1) - .element(0) - .extracting(TargetEventPublication::getEvent) - .isEqualTo(event); + if (completionMode == CompletionMode.DELETE) { + + assertThat(repository.findCompletedPublications()).isEmpty(); + assertThat(repository.findIncompletePublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } + } + + @Nested + @ContextConfiguration(classes = TestConfig.class) + @TestPropertySource(properties = CompletionMode.PROPERTY + "=DELETE") + static class WithDeleteCompletionTests extends JpaEventPublicationRepositoryIntegrationTests { + } private TargetEventPublication createPublication(Object event) { @@ -299,7 +325,7 @@ class JpaEventPublicationRepositoryIntegrationTests { em.persist(new JpaEventPublication(UUID.randomUUID(), date.toInstant(ZoneOffset.UTC), "", "", Object.class)); } - @Value + @lombok.Value private static final class TestEvent { String eventId; } diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationAutoConfiguration.java b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationAutoConfiguration.java index 67621dc7..9f8b26ca 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationAutoConfiguration.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationAutoConfiguration.java @@ -18,9 +18,11 @@ package org.springframework.modulith.events.mongodb; import org.springframework.boot.autoconfigure.AutoConfiguration; import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.context.annotation.Bean; +import org.springframework.core.env.Environment; import org.springframework.data.mongodb.core.MongoTemplate; import org.springframework.modulith.events.config.EventPublicationAutoConfiguration; import org.springframework.modulith.events.config.EventPublicationConfigurationExtension; +import org.springframework.modulith.events.support.CompletionMode; /** * Autoconfiguration for MongoDB event publication repository. @@ -32,7 +34,7 @@ import org.springframework.modulith.events.config.EventPublicationConfigurationE class MongoDbEventPublicationAutoConfiguration implements EventPublicationConfigurationExtension { @Bean - MongoDbEventPublicationRepository mongoDbEventPublicationRepository(MongoTemplate template) { - return new MongoDbEventPublicationRepository(template); + MongoDbEventPublicationRepository mongoDbEventPublicationRepository(MongoTemplate template, Environment environment) { + return new MongoDbEventPublicationRepository(template, CompletionMode.from(environment)); } } diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java index 0620f5ff..dad127ed 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java @@ -33,6 +33,7 @@ import org.springframework.data.util.TypeInformation; import org.springframework.modulith.events.core.EventPublicationRepository; import org.springframework.modulith.events.core.PublicationTargetIdentifier; import org.springframework.modulith.events.core.TargetEventPublication; +import org.springframework.modulith.events.support.CompletionMode; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; @@ -55,17 +56,21 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { private static final Sort DEFAULT_SORT = Sort.by(PUBLICATION_DATE).ascending(); private final MongoTemplate mongoTemplate; + private final CompletionMode completionMode; /** * Creates a new {@link MongoDbEventPublicationRepository} for the given {@link MongoTemplate}. * * @param mongoTemplate must not be {@literal null}. + * @param completionMode must not be {@literal null}. */ - public MongoDbEventPublicationRepository(MongoTemplate mongoTemplate) { + public MongoDbEventPublicationRepository(MongoTemplate mongoTemplate, CompletionMode completionMode) { Assert.notNull(mongoTemplate, "MongoTemplate must not be null!"); + Assert.notNull(completionMode, "Completion mode must not be null!"); this.mongoTemplate = mongoTemplate; + this.completionMode = completionMode; } /* @@ -87,9 +92,18 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { @Override public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) { - var update = Update.update(COMPLETION_DATE, completionDate); + var query = byEventAndListenerId(event, identifier); - mongoTemplate.findAndModify(byEventAndListenerId(event, identifier), update, MongoDbEventPublication.class); + if (completionMode == CompletionMode.DELETE) { + + mongoTemplate.remove(query, MongoDbEventPublication.class); + + } else { + + var update = Update.update(COMPLETION_DATE, completionDate); + + mongoTemplate.findAndModify(query, update, MongoDbEventPublication.class); + } } /* @@ -99,9 +113,18 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { @Override public void markCompleted(UUID identifier, Instant completionDate) { - var update = Update.update(COMPLETION_DATE, completionDate); + var criateria = query(where(ID).is(identifier)); - mongoTemplate.findAndModify(query(where(ID).is(identifier)), update, MongoDbEventPublication.class); + if (completionMode == CompletionMode.DELETE) { + + mongoTemplate.remove(criateria, MongoDbEventPublication.class); + + } else { + + var update = Update.update(COMPLETION_DATE, completionDate); + + mongoTemplate.findAndModify(criateria, update, MongoDbEventPublication.class); + } } /* diff --git a/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java b/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java index 4022ea2f..870ca599 100644 --- a/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java +++ b/spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java @@ -16,6 +16,7 @@ package org.springframework.modulith.events.mongodb; import static org.assertj.core.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.*; import lombok.Value; @@ -33,11 +34,14 @@ import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.test.autoconfigure.data.mongo.DataMongoTest; +import org.springframework.core.env.Environment; import org.springframework.data.mongodb.core.MongoTemplate; 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; /** * @author Björn Kieling @@ -51,12 +55,15 @@ class MongoDbEventPublicationRepositoryTest { private static final PublicationTargetIdentifier TARGET_IDENTIFIER = PublicationTargetIdentifier.of("listener"); @Autowired MongoTemplate mongoTemplate; + @Autowired Environment environment; MongoDbEventPublicationRepository repository; + CompletionMode completionMode; @BeforeEach void setUp() { - repository = new MongoDbEventPublicationRepository(mongoTemplate); + this.completionMode = CompletionMode.from(environment); + this.repository = new MongoDbEventPublicationRepository(mongoTemplate, completionMode); } @AfterEach @@ -137,11 +144,19 @@ class MongoDbEventPublicationRepositoryTest { repository.markCompleted(publication, Instant.now()); - assertThat(repository.findCompletedPublications()) - .hasSize(1) - .element(0) - .extracting(TargetEventPublication::getEvent) - .isEqualTo(event); + if (completionMode == CompletionMode.DELETE) { + + assertThat(repository.findCompletedPublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } + } @Test // GH-258 @@ -152,9 +167,17 @@ class MongoDbEventPublicationRepositoryTest { repository.markCompleted(publication.getIdentifier(), Instant.now()); - assertThat(repository.findCompletedPublications()) - .extracting(TargetEventPublication::getIdentifier) - .containsExactly(publication.getIdentifier()); + if (completionMode == CompletionMode.DELETE) { + + assertThat(repository.findCompletedPublications()).isEmpty(); + assertThat(repository.findIncompletePublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(publication.getIdentifier()); + } } private TargetEventPublication createPublication(Object event) { @@ -253,6 +276,8 @@ class MongoDbEventPublicationRepositoryTest { @Test // GH-251 void shouldDeleteCompletedEventsBefore() { + assumeTrue(completionMode == CompletionMode.UPDATE); + var first = createPublication(new TestEvent("abc")); var second = createPublication(new TestEvent("def")); @@ -283,6 +308,10 @@ class MongoDbEventPublicationRepositoryTest { } } + @Nested + @TestPropertySource(properties = CompletionMode.PROPERTY + "=DELETE") + static class WithDeleteCompletionTest extends MongoDbEventPublicationRepositoryTest {} + @Value private static final class TestEvent { String eventId; diff --git a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationAutoConfiguration.java b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationAutoConfiguration.java index 2702983d..2f526ae5 100644 --- a/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationAutoConfiguration.java +++ b/spring-modulith-events/spring-modulith-events-neo4j/src/main/java/org/springframework/modulith/events/neo4j/Neo4jEventPublicationAutoConfiguration.java @@ -6,10 +6,12 @@ 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.core.env.Environment; 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; +import org.springframework.modulith.events.support.CompletionMode; /** * Auto-configuration to register a {@link Neo4jEventPublicationRepository}, a default {@link Configuration} and a @@ -24,8 +26,9 @@ class Neo4jEventPublicationAutoConfiguration implements EventPublicationConfigur @Bean Neo4jEventPublicationRepository neo4jEventPublicationRepository(Neo4jClient neo4jClient, - Configuration cypherDslConfiguration, EventSerializer eventSerializer) { - return new Neo4jEventPublicationRepository(neo4jClient, cypherDslConfiguration, eventSerializer); + Configuration cypherDslConfiguration, EventSerializer eventSerializer, Environment environment) { + return new Neo4jEventPublicationRepository(neo4jClient, cypherDslConfiguration, eventSerializer, + CompletionMode.from(environment)); } @Bean 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 eacb1d3c..5b067cd0 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 @@ -38,6 +38,7 @@ 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.modulith.events.support.CompletionMode; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; import org.springframework.util.DigestUtils; @@ -71,6 +72,12 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .returning(EVENT_PUBLICATION_NODE) .build(); + private static final Statement DELETE_BY_EVENT_AND_LISTENER_ID = Cypher.match(EVENT_PUBLICATION_NODE) + .where(EVENT_PUBLICATION_NODE.property(EVENT_HASH).eq(Cypher.parameter(EVENT_HASH))) + .and(EVENT_PUBLICATION_NODE.property(LISTENER_ID).eq(Cypher.parameter(LISTENER_ID))) + .delete(EVENT_PUBLICATION_NODE) + .build(); + private static final Statement DELETE_BY_ID_STATEMENT = Cypher.match(EVENT_PUBLICATION_NODE) .where(EVENT_PUBLICATION_NODE.property(ID).in(Cypher.parameter(ID))) .delete(EVENT_PUBLICATION_NODE) @@ -131,17 +138,20 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { private final Neo4jClient neo4jClient; private final Renderer renderer; private final EventSerializer eventSerializer; + private final CompletionMode completionMode; Neo4jEventPublicationRepository(Neo4jClient neo4jClient, Configuration cypherDslConfiguration, - EventSerializer eventSerializer) { + EventSerializer eventSerializer, CompletionMode completionMode) { 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!"); + Assert.notNull(completionMode, "Completion mode must not be null!"); this.neo4jClient = neo4jClient; this.renderer = Renderer.getRenderer(cypherDslConfiguration); this.eventSerializer = eventSerializer; + this.completionMode = completionMode; } /* @@ -184,11 +194,21 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { var eventHash = DigestUtils.md5DigestAsHex(eventSerializer.serialize(event).toString().getBytes()); - neo4jClient.query(renderer.render(COMPLETE_STATEMENT)) - .bind(eventHash).to(EVENT_HASH) - .bind(identifier.getValue()).to(LISTENER_ID) - .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) - .run(); + if (completionMode == CompletionMode.DELETE) { + + neo4jClient.query(renderer.render(DELETE_BY_EVENT_AND_LISTENER_ID)) + .bind(eventHash).to(EVENT_HASH) + .bind(identifier.getValue()).to(LISTENER_ID) + .run(); + + } else { + + neo4jClient.query(renderer.render(COMPLETE_STATEMENT)) + .bind(eventHash).to(EVENT_HASH) + .bind(identifier.getValue()).to(LISTENER_ID) + .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) + .run(); + } } /* @@ -199,10 +219,17 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { @Transactional public void markCompleted(UUID identifier, Instant completionDate) { - neo4jClient.query(renderer.render(COMPLETE_BY_ID_STATEMENT)) - .bind(Values.value(identifier.toString())).to(ID) - .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) - .run(); + if (completionMode == CompletionMode.DELETE) { + + deletePublications(List.of(identifier)); + + } else { + + neo4jClient.query(renderer.render(COMPLETE_BY_ID_STATEMENT)) + .bind(Values.value(identifier.toString())).to(ID) + .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) + .run(); + } } /* 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 eec41bb3..e78bef27 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 @@ -1,6 +1,22 @@ +/* + * Copyright 2023-2024 the original author or authors. + * + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * + * https://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ package org.springframework.modulith.events.neo4j; import static org.assertj.core.api.Assertions.*; +import static org.junit.jupiter.api.Assumptions.*; import static org.mockito.Mockito.*; import lombok.Value; @@ -11,6 +27,7 @@ import java.util.List; import java.util.UUID; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; import org.neo4j.cypherdsl.core.renderer.Dialect; import org.neo4j.driver.AuthTokens; @@ -21,10 +38,13 @@ 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.core.env.Environment; 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.events.support.CompletionMode; import org.springframework.modulith.testapp.TestApplication; +import org.springframework.test.context.TestPropertySource; import org.springframework.test.context.junit.jupiter.SpringJUnitConfig; import org.springframework.util.DigestUtils; import org.testcontainers.containers.Neo4jContainer; @@ -47,11 +67,17 @@ class Neo4jEventPublicationRepositoryTest { @Autowired Neo4jEventPublicationRepository repository; @Autowired Driver driver; + @Autowired Environment environment; + @MockBean EventSerializer eventSerializer; + CompletionMode completionMode; + @BeforeEach void clearDb() { + this.completionMode = CompletionMode.from(environment); + try (var session = driver.session()) { session.run("MATCH (n) detach delete n").consume(); } @@ -189,6 +215,8 @@ class Neo4jEventPublicationRepositoryTest { @Test void deleteCompletedPublicationsBefore() throws Exception { + assumeTrue(completionMode == CompletionMode.UPDATE); + var testEvent1 = new TestEvent("id1"); var event1Serialized = "{\"eventId\":\"id1\"}"; var testEvent2 = new TestEvent("id2"); @@ -228,11 +256,19 @@ class Neo4jEventPublicationRepositoryTest { repository.markCompleted(publication, Instant.now()); - assertThat(repository.findCompletedPublications()) - .hasSize(1) - .element(0) - .extracting(TargetEventPublication::getEvent) - .isEqualTo(event); + if (completionMode == CompletionMode.DELETE) { + + assertThat(repository.findCompletedPublications()).isEmpty(); + assertThat(repository.findIncompletePublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } } @Test // GH-258 @@ -243,9 +279,17 @@ class Neo4jEventPublicationRepositoryTest { repository.markCompleted(publication.getIdentifier(), Instant.now()); - assertThat(repository.findCompletedPublications()) - .extracting(TargetEventPublication::getIdentifier) - .containsExactly(publication.getIdentifier()); + if (completionMode == CompletionMode.DELETE) { + + assertThat(repository.findCompletedPublications()).isEmpty(); + assertThat(repository.findIncompletePublications()).isEmpty(); + + } else { + + assertThat(repository.findCompletedPublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(publication.getIdentifier()); + } } private TargetEventPublication createPublication(Object event) { @@ -258,6 +302,10 @@ class Neo4jEventPublicationRepositoryTest { return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER)); } + @Nested + @TestPropertySource(properties = CompletionMode.PROPERTY + "=DELETE") + static class WithDeleteCompletionTest extends Neo4jEventPublicationRepositoryTest {} + @Value static class TestEvent { String eventId; diff --git a/src/docs/antora/modules/ROOT/pages/appendix.adoc b/src/docs/antora/modules/ROOT/pages/appendix.adoc index b911ba73..39dd221d 100644 --- a/src/docs/antora/modules/ROOT/pages/appendix.adoc +++ b/src/docs/antora/modules/ROOT/pages/appendix.adoc @@ -18,6 +18,10 @@ |The strategy to be applied to detect application modules. Can either be the class name of a custom implementation of `ApplicationModuleDetectionStrategy` or `direct-subpackages` (which is also the final fallback if nothing is configured) or `explicitly-annotated` to only select packages explicitly annotated with `@ApplicationModule` or jMolecules' `@Module`. See xref:fundamentals.adoc#customizing-modules[Customize Application Module Detection] for details. +|`spring.modulith.events.completion-mode` +|`UPDATE` +|How to mark event publications as completed. The default sets the completion date on the database entry, `DELETE` removes the entry right away. For details, see xref:events.adoc#publication-registry.completion[Event Publication Completion]. + |`spring.modulith.events.externalization.enabled` |`true` |Whether to enable event externalization. diff --git a/src/docs/antora/modules/ROOT/pages/events.adoc b/src/docs/antora/modules/ROOT/pages/events.adoc index a406193c..bf38d6f7 100644 --- a/src/docs/antora/modules/ROOT/pages/events.adoc +++ b/src/docs/antora/modules/ROOT/pages/events.adoc @@ -273,6 +273,20 @@ This artifact contains two primary abstractions that are available to applicatio * `CompletedEventPublications` -- This interface allows accessing all completed event publications, and provides an API to immediately purge all of them from the database or the completed publications older that a given duration (for example, 1 minute). * `IncompleteEventPublications` -- This interface allows accessing all incomplete event publications to resubmit either the ones matching a given predicate or older than a given `Duration` relative to the original publishing date. +[[publication-registry.completion]] +=== Event Publication Completion + +Event publications are marked as completed when a transactional or `@ApplicationModuleListener` execution completes successfully. +By default, the completion is registered by setting the completion date on an `EventPublication`. +This means that completed publications will remain in the Event Publication Registry so that they can be inspected through the `CompletedEventPublications` interface as described xref:events.adoc#publication-registry.managing-publications[above]. +A consequence of this is that you'll need to put some code in place that will periodically purge old, completed ``EventPublication``s. +Otherwise, the persistent abstraction of them, for example a relational database table, will grow unbounded and the interaction with the store creating and completing new ``EventPublication`` might slow down. + +Spring Modulith 1.3 introduces a configuration property `spring.modulith.events.completion-mode`. +It defaults to `UPDATE` which is backed by the strategy described above. +Alternatively, the completion mode can be set to `DELETE`, which alters the registry's persistence mechanisms to rather delete ``EventPublication``s on completion. +This means that `CompletedEventPublications` will not return any publications anymore, but at the same time, you don't have to worry about purging the completed events from the persistence store manually anymore. + [[publication-registry.publication-repositories]] === Event Publication Repositories