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