diff --git a/spring-modulith-events/spring-modulith-events-jdbc/pom.xml b/spring-modulith-events/spring-modulith-events-jdbc/pom.xml index 648841a6..9e7ff3af 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/pom.xml +++ b/spring-modulith-events/spring-modulith-events-jdbc/pom.xml @@ -80,6 +80,12 @@ test + + org.testcontainers + mysql + test + + org.hsqldb hsqldb @@ -98,6 +104,13 @@ test + + mysql + mysql-connector-java + 8.0.29 + test + + 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 92cd3583..2cc439ad 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 @@ -19,6 +19,8 @@ import java.io.IOException; import java.io.UncheckedIOException; import java.nio.charset.StandardCharsets; +import lombok.RequiredArgsConstructor; + import org.springframework.beans.factory.InitializingBean; import org.springframework.boot.jdbc.DatabaseDriver; import org.springframework.context.ResourceLoaderAware; @@ -35,34 +37,23 @@ import org.springframework.util.StreamUtils; * @author Björn Kieling * @author Oliver Drotbohm */ +@RequiredArgsConstructor class DatabaseSchemaInitializer implements ResourceLoaderAware, InitializingBean { private final JdbcTemplate jdbcTemplate; + private final DatabaseType databaseType; private ResourceLoader resourceLoader; - /** - * Creates a new {@link DatabaseSchemaInitializer} for the given {@link JdbcTemplate} and ini - * - * @param jdbcTemplate - * @param initEnabled - */ - public DatabaseSchemaInitializer(JdbcTemplate jdbcTemplate) { - - this.jdbcTemplate = jdbcTemplate; - } - @Override public void setResourceLoader(ResourceLoader resourceLoader) { this.resourceLoader = resourceLoader; } @Override - public void afterPropertiesSet() throws MetaDataAccessException { - - var fromDataSource = DatabaseDriver.fromDataSource(jdbcTemplate.getDataSource()); - var databaseName = fromDataSource.name().toLowerCase(); - var schemaDdlResource = resourceLoader.getResource("/schema-" + databaseName + ".sql"); + public void afterPropertiesSet() { + var schemaResourceFilename = databaseType.getSchemaResourceFilename(); + var schemaDdlResource = resourceLoader.getResource(schemaResourceFilename); var schemaDdl = asString(schemaDdlResource); jdbcTemplate.execute(schemaDdl); 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 new file mode 100644 index 00000000..fdd3eadb --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/DatabaseType.java @@ -0,0 +1,77 @@ +/* + * Copyright 2022 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 java.util.Map; +import java.util.UUID; + +import org.springframework.boot.jdbc.DatabaseDriver; + +/** + * @author Dmitry Belyaev + * @author Björn Kieling + */ +enum DatabaseType { + + HSQLDB("hsqldb"), + + H2("h2"), + + MYSQL("mysql") { + Object uuidToDatabase(UUID id) { + return id.toString(); + } + + UUID databaseToUUID(Object id) { + return UUID.fromString(id.toString()); + } + }, + + POSTGRES("postgresql"); + + private static final Map DATABASE_DRIVER_TO_DATABASE_TYPE_MAP = // + Map.of( // + DatabaseDriver.H2, H2, // + DatabaseDriver.HSQLDB, HSQLDB, // + DatabaseDriver.POSTGRESQL, POSTGRES, // + DatabaseDriver.MYSQL, MYSQL); + + static DatabaseType from(DatabaseDriver databaseDriver) { + var databaseType = DATABASE_DRIVER_TO_DATABASE_TYPE_MAP.get(databaseDriver); + if (databaseType == null) { + throw new IllegalArgumentException("Unsupported database type: " + databaseDriver); + } + return databaseType; + } + + private final String value; + + DatabaseType(String value) { + this.value = value; + } + + Object uuidToDatabase(UUID id) { + return id; + } + + UUID databaseToUUID(Object id) { + return (UUID) id; + } + + String getSchemaResourceFilename() { + return "/schema-" + value + ".sql"; + } +} 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 c76fcaa4..0be176a5 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 @@ -15,7 +15,10 @@ */ package org.springframework.modulith.events.jdbc; +import javax.sql.DataSource; + import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty; +import org.springframework.boot.jdbc.DatabaseDriver; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.jdbc.core.JdbcTemplate; @@ -31,13 +34,19 @@ import org.springframework.modulith.events.config.EventPublicationConfigurationE class JdbcEventPublicationAutoConfiguration implements EventPublicationConfigurationExtension { @Bean - JdbcEventPublicationRepository jdbcEventPublicationRepository(JdbcTemplate jdbcTemplate, EventSerializer serializer) { - return new JdbcEventPublicationRepository(jdbcTemplate, serializer); + DatabaseType databaseType(DataSource dataSource) { + var databaseDriver = DatabaseDriver.fromDataSource(dataSource); + return DatabaseType.from(databaseDriver); + } + @Bean + JdbcEventPublicationRepository jdbcEventPublicationRepository(JdbcTemplate jdbcTemplate, + EventSerializer serializer, DatabaseType databaseType) { + return new JdbcEventPublicationRepository(jdbcTemplate, serializer, databaseType); } @Bean @ConditionalOnProperty(name = "spring.modulith.events.schema-initialization.enabled", havingValue = "true") - DatabaseSchemaInitializer databaseSchemaInitializer(JdbcTemplate jdbcTemplate) { - return new DatabaseSchemaInitializer(jdbcTemplate); + DatabaseSchemaInitializer databaseSchemaInitializer(JdbcTemplate jdbcTemplate, DatabaseType databaseType) { + return new DatabaseSchemaInitializer(jdbcTemplate, databaseType); } } 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 d7b3dada..4a07586d 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 @@ -15,11 +15,6 @@ */ package org.springframework.modulith.events.jdbc; -import lombok.Builder; -import lombok.EqualsAndHashCode; -import lombok.RequiredArgsConstructor; -import lombok.extern.slf4j.Slf4j; - import java.sql.ResultSet; import java.sql.SQLException; import java.sql.Timestamp; @@ -40,6 +35,11 @@ import org.springframework.modulith.events.EventSerializer; import org.springframework.modulith.events.PublicationTargetIdentifier; import org.springframework.transaction.annotation.Transactional; +import lombok.Builder; +import lombok.EqualsAndHashCode; +import lombok.RequiredArgsConstructor; +import lombok.extern.slf4j.Slf4j; + /** * JDBC-based repository to store {@link EventPublication}s. * @@ -74,19 +74,23 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { ORDER BY PUBLICATION_DATE """; - private final JdbcOperations operations; + protected final JdbcOperations operations; private final EventSerializer serializer; + private final DatabaseType databaseType; + @Override @Transactional public EventPublication create(EventPublication publication) { - operations.update(SQL_STATEMENT_INSERT, // - UUID.randomUUID(), // + String serializedEvent = serializeEvent(publication.getEvent()); + operations.update( // + SQL_STATEMENT_INSERT, // + uuidToDatabase(UUID.randomUUID()), // publication.getEvent().getClass().getName(), // publication.getTargetIdentifier().getValue(), // Timestamp.from(publication.getPublicationDate()), // - serializeEvent(publication.getEvent())); + serializedEvent); return publication; } @@ -96,16 +100,16 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { public EventPublication update(CompletableEventPublication publication) { var serializedEvent = serializeEvent(publication.getEvent()); - var results = operations.query(SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, - (rs, rowNum) -> rs.getObject("ID", UUID.class), serializedEvent, publication.getTargetIdentifier().getValue()); + var listenerId = publication.getTargetIdentifier().getValue(); + var potentialPublicationIdsToBeUpdated = operations.query( // + SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, // + (rs, rowNum) -> getUUIDFromResultSet(rs), // + serializedEvent, // + listenerId); - if (!results.isEmpty()) { - - operations.update( // - SQL_STATEMENT_UPDATE, // - publication.getCompletionDate().map(Timestamp::from).orElse(null), // - results.get(0)); - } + potentialPublicationIdsToBeUpdated.stream() + .findFirst() + .ifPresent(id -> update(id, publication)); return publication; } @@ -115,17 +119,36 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { public Optional findIncompletePublicationsByEventAndTargetIdentifier( // Object event, PublicationTargetIdentifier targetIdentifier) { - var results = operations.query(SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, this::resultSetToPublications, - serializeEvent(event), targetIdentifier.getValue()); - - return Optional.ofNullable(results == null || results.isEmpty() ? null : results.get(0)); + String serializedEvent = serializeEvent(event); + String listenerId = targetIdentifier.getValue(); + return findAllIncompletePublicationsByEventAndListenerId(serializedEvent, listenerId).stream() // + .findFirst(); } @Override @Transactional(readOnly = true) @SuppressWarnings("null") public List findIncompletePublications() { - return operations.query(SQL_STATEMENT_FIND_UNCOMPLETED, this::resultSetToPublications); + return operations.query( // + SQL_STATEMENT_FIND_UNCOMPLETED, // + this::resultSetToPublications); + } + + private void update(UUID id, CompletableEventPublication publication) { + Timestamp timestamp = publication.getCompletionDate().map(Timestamp::from).orElse(null); + operations.update( // + SQL_STATEMENT_UPDATE, // + timestamp, // + uuidToDatabase(id)); + } + + private List findAllIncompletePublicationsByEventAndListenerId( + String serializedEvent, String listenerId) { + return operations.query( // + SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, // + this::resultSetToPublications, // + serializedEvent, // + listenerId); } private String serializeEvent(Object event) { @@ -142,11 +165,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { private List resultSetToPublications(ResultSet resultSet) throws SQLException { List result = new ArrayList<>(); - while (resultSet.next()) { - EventPublication publication = resultSetToPublication(resultSet); - if (publication != null) { result.add(publication); } @@ -165,17 +185,16 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { @Nullable private EventPublication resultSetToPublication(ResultSet rs) throws SQLException { - var id = rs.getObject("ID", UUID.class); - var eventClass = loadClass(id, rs.getString("EVENT_TYPE")); + UUID id = getUUIDFromResultSet(rs); + var eventClass = loadClass(id, rs.getString("EVENT_TYPE")); if (eventClass == null) { return null; } var completionDate = rs.getTimestamp("COMPLETION_DATE"); - return JdbcEventPublication.builder() - .completionDate(completionDate == null ? null : completionDate.toInstant()) + return JdbcEventPublication.builder().completionDate(completionDate == null ? null : completionDate.toInstant()) .eventType(eventClass) // .listenerId(rs.getString("LISTENER_ID")) // .publicationDate(rs.getTimestamp("PUBLICATION_DATE").toInstant()) // @@ -184,6 +203,15 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { .build(); } + private Object uuidToDatabase(UUID id) { + return databaseType.uuidToDatabase(id); + } + + private UUID getUUIDFromResultSet(ResultSet rs) throws SQLException { + Object id = rs.getObject("ID"); + return databaseType.databaseToUUID(id); + } + @Nullable private Class loadClass(UUID id, String className) { diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mysql.sql b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mysql.sql new file mode 100644 index 00000000..30a35c15 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/main/resources/schema-mysql.sql @@ -0,0 +1,10 @@ +CREATE TABLE IF NOT EXISTS EVENT_PUBLICATION +( + 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) +) diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerIntegrationTests.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerIntegrationTests.java index 3b24d604..358cdb9e 100644 --- a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerIntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseSchemaInitializerIntegrationTests.java @@ -114,4 +114,8 @@ class DatabaseSchemaInitializerIntegrationTests { @Nested @ActiveProfiles("postgres") class Postgres extends WithInitEnabled {} + + @Nested + @ActiveProfiles("mysql") + class MySQL extends WithInitEnabled {} } diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseTypeTest.java b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseTypeTest.java new file mode 100644 index 00000000..4c0ab875 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/DatabaseTypeTest.java @@ -0,0 +1,16 @@ +package org.springframework.modulith.events.jdbc; + +import static org.assertj.core.api.Assertions.*; + +import org.junit.jupiter.api.Test; +import org.springframework.boot.jdbc.DatabaseDriver; + +class DatabaseTypeTest { + + @Test + void shouldThrowExceptionOnUnsupportedDatabaseType() { + assertThatThrownBy(() -> DatabaseType.from(DatabaseDriver.UNKNOWN)) + .isInstanceOf(IllegalArgumentException.class) + .hasMessageContaining("UNKNOWN"); + } +} 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 6dc3e791..bb3effa1 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 @@ -217,6 +217,10 @@ class JdbcEventPublicationRepositoryIntegrationTests { @ActiveProfiles("postgres") class Postgres extends TestBase {} + @Nested + @ActiveProfiles("mysql") + class MySQL extends TestBase {} + @Value private static final class TestEvent { String eventId; diff --git a/spring-modulith-events/spring-modulith-events-jdbc/src/test/resources/application-mysql.properties b/spring-modulith-events/spring-modulith-events-jdbc/src/test/resources/application-mysql.properties new file mode 100644 index 00000000..555869e3 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-jdbc/src/test/resources/application-mysql.properties @@ -0,0 +1,6 @@ +spring.datasource.url=jdbc:tc:mysql:8.0.30:///events +spring.datasource.driverClassName=org.testcontainers.jdbc.ContainerDatabaseDriver + +spring.test.database.replace=NONE + +spring.modulith.events.schema-initialization.enabled=true