diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializer.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializer.java index 2a7d33be..56b8a145 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializer.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializer.java @@ -23,7 +23,6 @@ import org.springframework.beans.factory.InitializingBean; import org.springframework.core.io.ResourceLoader; import org.springframework.jdbc.core.JdbcOperations; import org.springframework.jdbc.datasource.init.ResourceDatabasePopulator; -import org.springframework.util.Assert; /** * Initializes the DB schema used to store events @@ -37,23 +36,15 @@ class DatabaseSchemaInitializer implements InitializingBean { private final DataSource dataSource; private final ResourceLoader resourceLoader; - private final DatabaseType databaseType; private final JdbcOperations jdbcOperations; - private final JdbcConfigurationProperties properties; + private final JdbcRepositorySettings settings; - DatabaseSchemaInitializer(DataSource dataSource, ResourceLoader resourceLoader, DatabaseType databaseType, - JdbcOperations jdbcOperations, JdbcConfigurationProperties properties) { - - Assert.isTrue(properties.getSchemaInitialization().isEnabled(), - "Schema initialization disabled! Initializer should not have been registered!"); + DatabaseSchemaInitializer(DataSource dataSource, ResourceLoader resourceLoader, JdbcOperations jdbcOperations, JdbcRepositorySettings settings) { this.dataSource = dataSource; this.resourceLoader = resourceLoader; - this.databaseType = databaseType; this.jdbcOperations = jdbcOperations; - this.properties = properties; - - properties.verify(databaseType); + this.settings = settings; } /* @@ -66,7 +57,8 @@ class DatabaseSchemaInitializer implements InitializingBean { try (Connection connection = dataSource.getConnection()) { var initialSchema = connection.getSchema(); - var schemaName = properties.getSchema(); + var schemaName = settings.getSchema(); + var databaseType = settings.getDatabaseType(); var useSchema = schemaName != null && !schemaName.isEmpty(); if (useSchema) { // A schema name has been specified. @@ -80,7 +72,10 @@ class DatabaseSchemaInitializer implements InitializingBean { } var locator = new DatabaseSchemaLocator(resourceLoader); - new ResourceDatabasePopulator(locator.getSchemaResource(databaseType)).execute(dataSource); + + var populator = new ResourceDatabasePopulator(); + locator.getSchemaResource(settings).forEach(populator::addScript); + populator.execute(dataSource); // Return to the initial schema. if (initialSchema != null && useSchema) { diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocator.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocator.java index 8a56ff20..5fd35ac7 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocator.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocator.java @@ -19,6 +19,9 @@ import org.springframework.core.io.Resource; import org.springframework.core.io.ResourceLoader; import org.springframework.util.Assert; +import java.util.Collection; +import java.util.List; + /** * Simple wrapper around a {@link ResourceLoader} to load database specific schema files from the classpath. * @@ -41,16 +44,22 @@ public class DatabaseSchemaLocator { } /** - * Loads the {@link Resource} containing the schema for the given {@link DatabaseType} from the classpath. + * Loads the {@link Resource} containing the schema for the given {@link JdbcRepositorySettings} from the classpath. * - * @param databaseType must not be {@literal null}. + * @param settings must not be {@literal null}. * @return will never be {@literal null}. */ - Resource getSchemaResource(DatabaseType databaseType) { + Collection getSchemaResource(JdbcRepositorySettings settings) { - Assert.notNull(databaseType, "DatabaseType must not be null!"); + Assert.notNull(settings, "JdbcRepositorySettings must not be null!"); + var databaseType = settings.getDatabaseType(); var schemaResourceFilename = databaseType.getSchemaResourceFilename(); - return resourceLoader.getResource(ResourceLoader.CLASSPATH_URL_PREFIX + schemaResourceFilename); + var schemaResource = resourceLoader.getResource(ResourceLoader.CLASSPATH_URL_PREFIX + schemaResourceFilename); + + return !settings.isArchiveCompletion() + ? List.of(schemaResource) + : List.of(schemaResource, resourceLoader.getResource(ResourceLoader.CLASSPATH_URL_PREFIX + databaseType.getArchiveSchemaResourceFilename())); + } } diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseType.java b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseType.java index 6d2c091d..3af37876 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseType.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseType.java @@ -25,6 +25,7 @@ import org.springframework.util.Assert; * @author Björn Kieling * @author Oliver Drotbohm * @author Raed Ben Hamouda + * @author Cora Iberkleid */ enum DatabaseType { @@ -106,7 +107,7 @@ enum DatabaseType { } }; - static final String SCHEMA_NOT_SUPPORTED = "Setting the schema name not supported on MySQL!"; + static final String SCHEMA_NOT_SUPPORTED = "Setting the schema name is not supported!"; static DatabaseType from(String productName) { @@ -138,6 +139,8 @@ enum DatabaseType { return "/schema-" + value + ".sql"; } + String getArchiveSchemaResourceFilename() { return "/schema-" + value + "-archive.sql"; } + boolean isSchemaSupported() { return true; } 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 29a9dd8a..4805f19f 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 @@ -19,6 +19,7 @@ import java.sql.DatabaseMetaData; import javax.sql.DataSource; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.boot.autoconfigure.AutoConfigureBefore; import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; import org.springframework.boot.context.properties.EnableConfigurationProperties; @@ -44,6 +45,8 @@ import org.springframework.modulith.events.support.CompletionMode; @EnableConfigurationProperties(JdbcConfigurationProperties.class) class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigurationExtension { + @Autowired Environment environment; + @Bean DatabaseType databaseType(DataSource dataSource) { return DatabaseType.from(fromDataSource(dataSource)); @@ -51,7 +54,7 @@ class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigura @Bean JdbcRepositorySettings jdbcEventPublicationRepositorySettings(DatabaseType databaseType, - JdbcConfigurationProperties properties, Environment environment) { + JdbcConfigurationProperties properties) { return new JdbcRepositorySettings(databaseType, CompletionMode.from(environment), properties.getSchema()); } @@ -65,9 +68,9 @@ class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigura @Bean @ConditionalOnProperty(name = "spring.modulith.events.jdbc.schema-initialization.enabled", havingValue = "true") DatabaseSchemaInitializer databaseSchemaInitializer(DataSource dataSource, ResourceLoader resourceLoader, - DatabaseType databaseType, JdbcTemplate jdbcTemplate, JdbcConfigurationProperties properties) { + DatabaseType databaseType, JdbcTemplate jdbcTemplate, JdbcRepositorySettings settings) { - return new DatabaseSchemaInitializer(dataSource, resourceLoader, databaseType, jdbcTemplate, properties); + return new DatabaseSchemaInitializer(dataSource, resourceLoader, jdbcTemplate, settings); } private static String fromDataSource(DataSource dataSource) { 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 25cfa455..34413941 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 @@ -50,6 +50,7 @@ import org.springframework.util.ObjectUtils; * @author Björn Kieling * @author Oliver Drotbohm * @author Raed Ben Hamouda + * @author Cora Iberkleid */ class JdbcEventPublicationRepository implements EventPublicationRepository, BeanClassLoaderAware { @@ -130,20 +131,39 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean ID = ? """; - private static final String SQL_STATEMENT_DELETE_UNCOMPLETED = """ + private static final String SQL_STATEMENT_DELETE_COMPLETED = """ DELETE FROM %s WHERE COMPLETION_DATE IS NOT NULL """; - private static final String SQL_STATEMENT_DELETE_UNCOMPLETED_BEFORE = """ + private static final String SQL_STATEMENT_DELETE_COMPLETED_BEFORE = """ DELETE FROM %s WHERE COMPLETION_DATE < ? """; + private static final String SQL_STATEMENT_COPY_TO_ARCHIVE_BY_ID = """ + -- Only copy if no entry in target table + INSERT INTO %s (ID, LISTENER_ID, EVENT_TYPE, SERIALIZED_EVENT, PUBLICATION_DATE, COMPLETION_DATE) + SELECT ID, LISTENER_ID, EVENT_TYPE, SERIALIZED_EVENT, PUBLICATION_DATE, ? + FROM %s + WHERE ID = ? + AND NOT EXISTS (SELECT 1 FROM %s WHERE ID = EVENT_PUBLICATION.ID) + """; + + private static final String SQL_STATEMENT_COPY_TO_ARCHIVE_BY_EVENT_AND_LISTENER_ID = """ + -- Only copy if no entry in target table + INSERT INTO %s (ID, LISTENER_ID, EVENT_TYPE, SERIALIZED_EVENT, PUBLICATION_DATE, COMPLETION_DATE) + SELECT ID, LISTENER_ID, EVENT_TYPE, SERIALIZED_EVENT, PUBLICATION_DATE, ? + FROM %s + WHERE LISTENER_ID = ? + AND SERIALIZED_EVENT = ? + AND NOT EXISTS (SELECT 1 FROM %s WHERE ID = EVENT_PUBLICATION.ID) + """; + private static final int DELETE_BATCH_SIZE = 100; private final JdbcOperations operations; @@ -162,8 +182,10 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean sqlStatementDelete, sqlStatementDeleteByEventAndListenerId, sqlStatementDeleteById, - sqlStatementDeleteUncompleted, - sqlStatementDeleteUncompletedBefore; + sqlStatementDeleteCompleted, + sqlStatementDeleteCompletedBefore, + sqlStatementCopyToArchive, + sqlStatementCopyToArchiveByEventAndListenerId; /** * Creates a new {@link JdbcEventPublicationRepository} for the given {@link JdbcOperations}, {@link EventSerializer}, @@ -186,9 +208,10 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean var schema = settings.getSchema(); var table = ObjectUtils.isEmpty(schema) ? "EVENT_PUBLICATION" : schema + ".EVENT_PUBLICATION"; + var completedTable = settings.isArchiveCompletion() ? table + "_ARCHIVE" : table; this.sqlStatementInsert = SQL_STATEMENT_INSERT.formatted(table); - this.sqlStatementFindCompleted = SQL_STATEMENT_FIND_COMPLETED.formatted(table); + this.sqlStatementFindCompleted = SQL_STATEMENT_FIND_COMPLETED.formatted(completedTable); this.sqlStatementFindUncompleted = SQL_STATEMENT_FIND_UNCOMPLETED.formatted(table); this.sqlStatementFindUncompletedBefore = SQL_STATEMENT_FIND_UNCOMPLETED_BEFORE.formatted(table); this.sqlStatementUpdateByEventAndListenerId = SQL_STATEMENT_UPDATE_BY_EVENT_AND_LISTENER_ID.formatted(table); @@ -197,8 +220,10 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean 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); + this.sqlStatementDeleteCompleted = SQL_STATEMENT_DELETE_COMPLETED.formatted(completedTable); + this.sqlStatementDeleteCompletedBefore = SQL_STATEMENT_DELETE_COMPLETED_BEFORE.formatted(completedTable); + this.sqlStatementCopyToArchive = SQL_STATEMENT_COPY_TO_ARCHIVE_BY_ID.formatted(completedTable, table, completedTable); + this.sqlStatementCopyToArchiveByEventAndListenerId = SQL_STATEMENT_COPY_TO_ARCHIVE_BY_EVENT_AND_LISTENER_ID.formatted(completedTable, table, completedTable); } /* @@ -246,6 +271,14 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean operations.update(sqlStatementDeleteByEventAndListenerId, targetIdentifier, serializedEvent); + } else if (settings.isArchiveCompletion()) { + + operations.update(sqlStatementCopyToArchiveByEventAndListenerId, // + Timestamp.from(completionDate), // + targetIdentifier, // + serializedEvent); + operations.update(sqlStatementDeleteByEventAndListenerId, targetIdentifier, serializedEvent); + } else { operations.update(sqlStatementUpdateByEventAndListenerId, // @@ -263,10 +296,18 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean @Transactional public void markCompleted(UUID identifier, Instant completionDate) { + var databaseId = uuidToDatabase(identifier); + var timestamp = Timestamp.from(completionDate); + if (settings.isDeleteCompletion()) { - operations.update(sqlStatementDeleteById, uuidToDatabase(identifier)); + operations.update(sqlStatementDeleteById, databaseId); + + } else if (settings.isArchiveCompletion()) { + operations.update(sqlStatementCopyToArchive, timestamp, databaseId); + operations.update(sqlStatementDeleteById, databaseId); + } else { - operations.update(sqlStatementUpdateById, Timestamp.from(completionDate), uuidToDatabase(identifier)); + operations.update(sqlStatementUpdateById, timestamp, databaseId); } } @@ -338,7 +379,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean */ @Override public void deleteCompletedPublications() { - operations.execute(sqlStatementDeleteUncompleted); + operations.execute(sqlStatementDeleteCompleted); } /* @@ -350,7 +391,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean Assert.notNull(instant, "Instant must not be null!"); - operations.update(sqlStatementDeleteUncompletedBefore, Timestamp.from(instant)); + operations.update(sqlStatementDeleteCompletedBefore, Timestamp.from(instant)); } private String serializeEvent(Object event) { @@ -457,9 +498,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean * @param id must not be {@literal null}. * @param publicationDate must not be {@literal null}. * @param listenerId must not be {@literal null} or empty. - * @param serializedEvent must not be {@literal null} or empty. - * @param eventType must not be {@literal null}. - * @param serializer must not be {@literal null}. + * @param event must not be {@literal null}.. * @param completionDate can be {@literal null}. */ public JdbcEventPublication(UUID id, Instant publicationDate, String listenerId, Supplier event, @@ -468,6 +507,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean Assert.notNull(id, "Id must not be null!"); Assert.notNull(publicationDate, "Publication date must not be null!"); Assert.hasText(listenerId, "Listener id must not be null or empty!"); + Assert.notNull(event, "Event must not be null!"); this.id = id; this.publicationDate = publicationDate; 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 index a297a954..99b5ee74 100644 --- 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 @@ -47,6 +47,10 @@ public class JdbcRepositorySettings { this.databaseType = databaseType; this.schema = schema; this.completionMode = completionMode; + + if (schema != null && !databaseType.isSchemaSupported()) { + throw new IllegalStateException(DatabaseType.SCHEMA_NOT_SUPPORTED); + } } /** @@ -74,4 +78,14 @@ public class JdbcRepositorySettings { public boolean isDeleteCompletion() { return completionMode == CompletionMode.DELETE; } + + /** + * Returns whether we use the archiving completion mode. + */ + public boolean isArchiveCompletion() { return completionMode == CompletionMode.ARCHIVE; } + + /** + * Returns whether we use the updating completion mode. + */ + public boolean isUpdateCompletion() { return completionMode == CompletionMode.UPDATE; } } diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-h2-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-h2-archive.sql new file mode 100644 index 00000000..a9ded947 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-h2-archive.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE +( + ID UUID NOT NULL, + COMPLETION_DATE TIMESTAMP(9) WITH TIME ZONE, + EVENT_TYPE VARCHAR(512) NOT NULL, + LISTENER_ID VARCHAR(512) NOT NULL, + PUBLICATION_DATE TIMESTAMP(9) WITH TIME ZONE NOT NULL, + SERIALIZED_EVENT VARCHAR(4000) NOT NULL, + PRIMARY KEY (ID) + ); +CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE_BY_LISTENER_ID_AND_SERIALIZED_EVENT_IDX ON EVENT_PUBLICATION_ARCHIVE (LISTENER_ID, SERIALIZED_EVENT); +CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE_BY_COMPLETION_DATE_IDX ON EVENT_PUBLICATION_ARCHIVE (COMPLETION_DATE); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-hsqldb-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-hsqldb-archive.sql new file mode 100644 index 00000000..4fe4510d --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-hsqldb-archive.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE +( + ID UUID NOT NULL, + COMPLETION_DATE TIMESTAMP(9), + EVENT_TYPE VARCHAR(512) NOT NULL, + LISTENER_ID VARCHAR(512) NOT NULL, + PUBLICATION_DATE TIMESTAMP(9) NOT NULL, + SERIALIZED_EVENT VARCHAR(4000) NOT NULL, + PRIMARY KEY (ID) + ); +CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE_BY_LISTENER_ID_AND_SERIALIZED_EVENT_IDX ON EVENT_PUBLICATION_ARCHIVE (LISTENER_ID, SERIALIZED_EVENT); +CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE_BY_COMPLETION_DATE_IDX ON EVENT_PUBLICATION_ARCHIVE (COMPLETION_DATE); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mariadb-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mariadb-archive.sql new file mode 100644 index 00000000..ca997665 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mariadb-archive.sql @@ -0,0 +1,11 @@ +CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE +( + ID VARCHAR(36) NOT NULL, + LISTENER_ID VARCHAR(512) NOT NULL, + EVENT_TYPE VARCHAR(512) NOT NULL, + SERIALIZED_EVENT VARCHAR(4000) NOT NULL, + PUBLICATION_DATE TIMESTAMP(6) NOT NULL, + COMPLETION_DATE TIMESTAMP(6) DEFAULT NULL NULL, + PRIMARY KEY (ID), + INDEX EVENT_PUBLICATION_ARCHIVE_BY_COMPLETION_DATE_IDX (COMPLETION_DATE) + ); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mysql-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mysql-archive.sql new file mode 100644 index 00000000..ca997665 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mysql-archive.sql @@ -0,0 +1,11 @@ +CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE +( + ID VARCHAR(36) NOT NULL, + LISTENER_ID VARCHAR(512) NOT NULL, + EVENT_TYPE VARCHAR(512) NOT NULL, + SERIALIZED_EVENT VARCHAR(4000) NOT NULL, + PUBLICATION_DATE TIMESTAMP(6) NOT NULL, + COMPLETION_DATE TIMESTAMP(6) DEFAULT NULL NULL, + PRIMARY KEY (ID), + INDEX EVENT_PUBLICATION_ARCHIVE_BY_COMPLETION_DATE_IDX (COMPLETION_DATE) + ); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle-archive.sql new file mode 100644 index 00000000..7e783444 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle-archive.sql @@ -0,0 +1,12 @@ +-- oracle database support 'create xx if not exists' from 23c, lower version should create table and index manually +CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE ( + ID VARCHAR2(36) NOT NULL, + LISTENER_ID VARCHAR2(512) NOT NULL, + EVENT_TYPE VARCHAR2(512) NOT NULL, + SERIALIZED_EVENT VARCHAR2(4000) NOT NULL, + PUBLICATION_DATE TIMESTAMP(6) NOT NULL, + COMPLETION_DATE TIMESTAMP(6), + CONSTRAINT EVENT_PUBLICATION_ARCHIVE_PK PRIMARY KEY(ID) + ); + CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE_BY_LISTENER_ID_AND_SERIALIZED_EVENT_IDX ON EVENT_PUBLICATION_ARCHIVE (LISTENER_ID, SERIALIZED_EVENT); + CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_ARCHIVE_BY_COMPLETION_DATE_IDX ON EVENT_PUBLICATION_ARCHIVE (COMPLETION_DATE); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle.sql index d203fb71..1258783f 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle.sql +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-oracle.sql @@ -8,4 +8,5 @@ CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION ( COMPLETION_DATE TIMESTAMP(6), CONSTRAINT EVENT_PUBLICATION_PK PRIMARY KEY(ID) ); +CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_BY_LISTENER_ID_AND_SERIALIZED_EVENT_IDX ON EVENT_PUBLICATION (LISTENER_ID, SERIALIZED_EVENT); CREATE INDEX IF NOT EXISTS EVENT_PUBLICATION_BY_COMPLETION_DATE_IDX ON EVENT_PUBLICATION (COMPLETION_DATE); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-postgresql-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-postgresql-archive.sql new file mode 100644 index 00000000..4a0d264c --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-postgresql-archive.sql @@ -0,0 +1,12 @@ +CREATE TABLE IF NOT EXISTS event_publication_archive +( + id UUID NOT NULL, + listener_id TEXT NOT NULL, + event_type TEXT NOT NULL, + serialized_event TEXT NOT NULL, + publication_date TIMESTAMP WITH TIME ZONE NOT NULL, + completion_date TIMESTAMP WITH TIME ZONE, + PRIMARY KEY (id) + ); +CREATE INDEX IF NOT EXISTS event_publication_archive_serialized_event_hash_idx ON event_publication_archive USING hash(serialized_event); +CREATE INDEX IF NOT EXISTS event_publication_archive_by_completion_date_idx ON event_publication_archive (completion_date); diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-sqlserver-archive.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-sqlserver-archive.sql new file mode 100644 index 00000000..6b28118f --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-sqlserver-archive.sql @@ -0,0 +1,12 @@ +IF NOT EXISTS(SELECT * FROM INFORMATION_SCHEMA.TABLES WHERE TABLE_NAME = 'EVENT_PUBLICATION_ARCHIVE') +CREATE TABLE EVENT_PUBLICATION_ARCHIVE +( + ID VARCHAR(36) NOT NULL, + LISTENER_ID VARCHAR(512) NOT NULL, + EVENT_TYPE VARCHAR(512) NOT NULL, + SERIALIZED_EVENT VARCHAR(MAX) NOT NULL, + PUBLICATION_DATE DATETIME2(6) NOT NULL, + COMPLETION_DATE DATETIME2(6) NULL, + PRIMARY KEY (ID), + INDEX EVENT_PUBLICATION_ARCHIVE_BY_COMPLETION_DATE_IDX (COMPLETION_DATE) +); \ No newline at end of file diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerUnitTests.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerUnitTests.java index cb5b177a..a44bf066 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerUnitTests.java @@ -27,6 +27,7 @@ import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.core.io.ResourceLoader; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.modulith.events.jdbc.JdbcConfigurationProperties.SchemaInitialization; +import org.springframework.modulith.events.support.CompletionMode; /** * Unit tests for {@link DatabaseSchemaInitializer}. @@ -50,7 +51,7 @@ class DatabaseSchemaInitializerUnitTests { // GH-836 @ParameterizedTest @ValueSource(strings = { "", "test" }) - void rejectsExplicitSchemaNameForMariDB(String schema) { + void rejectsExplicitSchemaNameForMariaDB(String schema) { assertThatIllegalStateException().isThrownBy(() -> createInitializer(withSchema(schema), DatabaseType.MARIADB)); } @@ -61,8 +62,17 @@ class DatabaseSchemaInitializerUnitTests { assertThatIllegalStateException().isThrownBy(() -> createInitializer(withSchema(schema), DatabaseType.MSSQL)); } + @ParameterizedTest + @ValueSource(strings = { "", "test" }) + void rejectsExplicitSchemaNameForOracle(String schema) { + assertThatIllegalStateException().isThrownBy(() -> createInitializer(withSchema(schema), DatabaseType.ORACLE)); + } + private DatabaseSchemaInitializer createInitializer(JdbcConfigurationProperties properties, DatabaseType type) { - return new DatabaseSchemaInitializer(dataSource, resourceLoader, type, jdbcTemplate, properties); + + var settings = new JdbcRepositorySettings(type, CompletionMode.UPDATE, properties.getSchema()); + + return new DatabaseSchemaInitializer(dataSource, resourceLoader, jdbcTemplate, settings); } private static JdbcConfigurationProperties withSchema(String schema) { diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocatorUnitTests.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocatorUnitTests.java index f20931e8..1b453f9f 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocatorUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaLocatorUnitTests.java @@ -26,6 +26,7 @@ import org.mockito.Mock; import org.mockito.junit.jupiter.MockitoExtension; import org.springframework.core.io.ClassPathResource; import org.springframework.core.io.ResourceLoader; +import org.springframework.modulith.events.support.CompletionMode; /** * Unit tests for {@link DatabaseSchemaInitializer}. @@ -46,8 +47,9 @@ public class DatabaseSchemaLocatorUnitTests { var locator = new DatabaseSchemaLocator(resourceLoader); var captor = ArgumentCaptor.forClass(String.class); + var settings = new JdbcRepositorySettings(DatabaseType.H2, CompletionMode.UPDATE, null); - assertThatNoException().isThrownBy(() -> locator.getSchemaResource(DatabaseType.H2)); + assertThatNoException().isThrownBy(() -> locator.getSchemaResource(settings)); verify(resourceLoader).getResource(captor.capture()); assertThat(captor.getValue()).startsWith(ResourceLoader.CLASSPATH_URL_PREFIX); } 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 74ff9f9b..e2650fca 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 @@ -29,6 +29,7 @@ import java.time.temporal.ChronoUnit; import java.util.Comparator; import java.util.List; +import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Nested; import org.junit.jupiter.api.Test; @@ -53,6 +54,7 @@ import org.testcontainers.junit.jupiter.Testcontainers; * @author Björn Kieling * @author Oliver Drotbohm * @author Raed Ben Hamouda + * @author Cora Iberkleid */ class JdbcEventPublicationRepositoryIntegrationTests { @@ -69,9 +71,15 @@ class JdbcEventPublicationRepositoryIntegrationTests { @MockitoBean EventSerializer serializer; + @AfterEach @BeforeEach void cleanUp() { + operations.execute("TRUNCATE TABLE " + table()); + + if (properties.isArchiveCompletion()) { + operations.execute("TRUNCATE TABLE " + archiveTable()); + } } @Test // GH-3 @@ -228,13 +236,20 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); var publication = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER)); - repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER)); + repository.markCompleted(publication, Instant.now()); + repository.deleteCompletedPublications(); assertThat(operations.query("SELECT * FROM " + table(), (rs, __) -> rs.getString("SERIALIZED_EVENT"))) .hasSize(1).element(0).isEqualTo(serializedEvent2); + + if (properties.isArchiveCompletion()) { + assertThat(operations.query("SELECT * FROM " + archiveTable(), (rs, __) -> rs.getString("SERIALIZED_EVENT"))) + .hasSize(0); + } + } @Test // GH-251 @@ -259,9 +274,12 @@ class JdbcEventPublicationRepositoryIntegrationTests { repository.markCompleted(testEvent1, TARGET_IDENTIFIER, now.minusSeconds(30)); repository.markCompleted(testEvent2, TARGET_IDENTIFIER, now); + repository.deleteCompletedPublicationsBefore(now.minusSeconds(15)); - assertThat(operations.query("SELECT * FROM " + table(), (rs, __) -> rs.getString("SERIALIZED_EVENT"))) + var table = properties.isArchiveCompletion() ? archiveTable() : table(); + + assertThat(operations.query("SELECT * FROM " + table, (rs, __) -> rs.getString("SERIALIZED_EVENT"))) .hasSize(1).element(0).isEqualTo(serializedEvent2); } @@ -331,9 +349,10 @@ class JdbcEventPublicationRepositoryIntegrationTests { repository.markCompleted(publication.getIdentifier(), Instant.now()); + assertThat(repository.findIncompletePublications()).isEmpty(); + if (properties.isDeleteCompletion()) { - assertThat(repository.findIncompletePublications()).isEmpty(); assertThat(repository.findCompletedPublications()).isEmpty(); } else { @@ -342,6 +361,10 @@ class JdbcEventPublicationRepositoryIntegrationTests { .extracting(TargetEventPublication::getIdentifier) .containsExactly(publication.getIdentifier()); } + + if (properties.isArchiveCompletion()) { + assertThat(operations.queryForObject("SELECT COUNT(*) FROM " + archiveTable(), int.class)).isOne(); + } } @Test // GH-753 @@ -367,6 +390,8 @@ class JdbcEventPublicationRepositoryIntegrationTests { return "EVENT_PUBLICATION"; } + String archiveTable() { return table() + "_ARCHIVE"; } + private TargetEventPublication createPublication(Object event) { var token = event.toString(); @@ -378,36 +403,36 @@ class JdbcEventPublicationRepositoryIntegrationTests { } } - @Nested @JdbcTest(properties = "spring.modulith.events.jdbc.schema-initialization.enabled=true") - static class WithNoDefinedSchemaName extends TestBase {} + static abstract class WithNoDefinedSchemaName extends TestBase {} - @Nested @JdbcTest(properties = { "spring.modulith.events.jdbc.schema-initialization.enabled=true", "spring.modulith.events.jdbc.schema=test" }) - static class WithDefinedSchemaName extends TestBase { + static abstract class WithDefinedSchemaName extends TestBase { @Override String table() { - return "test.EVENT_PUBLICATION"; + return "test." + super.table(); } } - @Nested @JdbcTest(properties = { "spring.modulith.events.jdbc.schema-initialization.enabled=true", "spring.modulith.events.jdbc.schema=" }) - static class WithEmptySchemaName extends TestBase { + static abstract class WithEmptySchemaName extends TestBase {} - @Override - String table() { - return "EVENT_PUBLICATION"; - } - } - - @Nested @JdbcTest(properties = { "spring.modulith.events.jdbc.schema-initialization.enabled=true", CompletionMode.PROPERTY + "=DELETE" }) - static class WithDeleteCompletion extends TestBase {} + static abstract class WithDeleteCompletion extends TestBase {} + + @JdbcTest(properties = { "spring.modulith.events.jdbc.schema-initialization.enabled=true", + CompletionMode.PROPERTY + "=ARCHIVE" }) + static abstract class WithArchiveCompletion extends TestBase { + + @Override + String archiveTable() { + return "EVENT_PUBLICATION_ARCHIVE"; + } + } // HSQL @@ -421,7 +446,10 @@ class JdbcEventPublicationRepositoryIntegrationTests { class HsqlWithEmptySchemaName extends WithEmptySchemaName {} @WithHsql - class HsqlWithEmptyDeleteCompletion extends WithDeleteCompletion {} + class HsqlWithDeleteCoqmpletion extends WithDeleteCompletion {} + + @WithHsql + class HsqlWithArchiveCompletion extends WithArchiveCompletion {} // H2 @@ -435,7 +463,10 @@ class JdbcEventPublicationRepositoryIntegrationTests { class H2WithEmptySchemaName extends WithEmptySchemaName {} @WithH2 - class H2WithEmptyDeleteCompletion extends WithDeleteCompletion {} + class H2WithDeleteCompletion extends WithDeleteCompletion {} + + @WithH2 + class H2WithArchiveCompletion extends WithArchiveCompletion {} // Postgres @@ -451,13 +482,8 @@ class JdbcEventPublicationRepositoryIntegrationTests { @WithPostgres class PostgresWithDeleteCompletion extends WithDeleteCompletion {} - // MSSQL - - @WithMssql - class MssqlWithNoDefinedSchemaName extends WithNoDefinedSchemaName {} - - @WithMssql - class MssqlWithDeleteCompletion extends WithDeleteCompletion {} + @WithPostgres + class PostgresWithArchiveCompletion extends WithArchiveCompletion {} // MySQL @@ -467,6 +493,9 @@ class JdbcEventPublicationRepositoryIntegrationTests { @WithMySql class MysqlWithDeleteCompletion extends WithDeleteCompletion {} + @WithMySql + class MysqlWithArchiveCompletion extends WithArchiveCompletion {} + // MariaDB @WithMariaDB @@ -475,6 +504,20 @@ class JdbcEventPublicationRepositoryIntegrationTests { @WithMariaDB class MariaDBWithDeleteCompletion extends WithDeleteCompletion {} + @WithMariaDB + class MariaDBWithArchiveCompletion extends WithArchiveCompletion {} + + // MSSQL + + @WithMssql + class MssqlWithNoDefinedSchemaName extends WithNoDefinedSchemaName {} + + @WithMssql + class MssqlWithDeleteCompletion extends WithDeleteCompletion {} + + @WithMssql + class MssqlWithArchiveCompletion extends WithArchiveCompletion {} + // Oracle @WithOracle @@ -483,6 +526,9 @@ class JdbcEventPublicationRepositoryIntegrationTests { @WithOracle class OracleWithDeleteCompletion extends WithDeleteCompletion {} + @WithOracle + class OracleWithArchiveCompletion extends WithArchiveCompletion {} + private record TestEvent(String eventId) {} private static final class Sample {}