diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublicationRepository.java index f214fcbd..b858f1d5 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/EventPublicationRepository.java @@ -59,4 +59,9 @@ public interface EventPublicationRepository { */ Optional findIncompletePublicationsByEventAndTargetIdentifier( // Object event, PublicationTargetIdentifier targetIdentifier); + + /** + * Deletes all publications that were already marked as completed. + */ + void deleteCompletedPublications(); } 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 6c879ec5..c807b8aa 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 @@ -66,13 +66,21 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { WHERE ID = ? """; private static final String SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID = """ - SELECT * FROM EVENT_PUBLICATION + SELECT * + FROM EVENT_PUBLICATION WHERE SERIALIZED_EVENT = ? AND LISTENER_ID = ? AND COMPLETION_DATE IS NULL ORDER BY PUBLICATION_DATE """; + private static final String SQL_STATEMENT_DELETE_UNCOMPLETED = """ + DELETE + FROM EVENT_PUBLICATION + WHERE + COMPLETION_DATE IS NOT NULL + """; + private final JdbcOperations operations; private final EventSerializer serializer; @@ -136,6 +144,11 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { this::resultSetToPublications); } + @Override + public void deleteCompletedPublications() { + operations.execute(SQL_STATEMENT_DELETE_UNCOMPLETED); + } + private void update(UUID id, CompletableEventPublication publication) { var timestamp = publication.getCompletionDate().map(Timestamp::from).orElse(null); 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 bb3effa1..9f630cb4 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 @@ -203,6 +203,39 @@ class JdbcEventPublicationRepositoryIntegrationTests { .isEmpty(); } } + + @Nested + class DeleteCompletedPublications { + + @Test + // GH-20 + void shouldDeleteCompletedEvents() { + TestEvent testEvent1 = new TestEvent("abc"); + String serializedEvent1 = "{\"eventId\":\"abc\"}"; + TestEvent testEvent2 = new TestEvent("def"); + String serializedEvent2 = "{\"eventId\":\"def\"}"; + + when(serializer.serialize(testEvent1)).thenReturn(serializedEvent1); + when(serializer.deserialize(serializedEvent1, TestEvent.class)).thenReturn(testEvent1); + when(serializer.serialize(testEvent2)).thenReturn(serializedEvent2); + when(serializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); + + CompletableEventPublication publication1 = CompletableEventPublication.of(testEvent1, TARGET_IDENTIFIER); + CompletableEventPublication publication2 = CompletableEventPublication.of(testEvent2, TARGET_IDENTIFIER); + + repository.create(publication1); + repository.create(publication2); + + repository.update(publication1.markCompleted()); + + repository.deleteCompletedPublications(); + + var eventPublications = operations.query( + "SELECT * FROM EVENT_PUBLICATION", (rs, __) -> rs.getString("SERIALIZED_EVENT")); + assertThat(eventPublications).hasSize(1); + assertThat(eventPublications.get(0)).isEqualTo(serializedEvent2); + } + } } @Nested 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 ddc30201..c989e7e5 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 @@ -39,7 +39,6 @@ import org.springframework.transaction.annotation.Transactional; */ @RequiredArgsConstructor public class JpaEventPublicationRepository implements EventPublicationRepository { - private static String BY_EVENT_AND_LISTENER_ID = """ select p from JpaEventPublication p @@ -56,6 +55,13 @@ public class JpaEventPublicationRepository implements EventPublicationRepository p.completionDate is null """; + private static final String DELETE_COMPLETED = """ + delete + from JpaEventPublication p + where + p.completionDate is not null + """; + private final EntityManager entityManager; private final EventSerializer serializer; @@ -100,6 +106,12 @@ public class JpaEventPublicationRepository implements EventPublicationRepository .map(this::entityToDomain); } + @Override + @Transactional + public void deleteCompletedPublications() { + entityManager.createQuery(DELETE_COMPLETED).executeUpdate(); + } + private Optional findEntityBySerializedEventAndListenerIdAndCompletionDateNull( // Object event, String listenerId) { 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 ec92c900..26de0177 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 @@ -18,17 +18,19 @@ package org.springframework.modulith.events.jpa; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; -import jakarta.persistence.EntityManager; -import jakarta.persistence.EntityManagerFactory; import lombok.RequiredArgsConstructor; import lombok.Value; +import jakarta.persistence.EntityManager; +import jakarta.persistence.EntityManagerFactory; + import java.util.List; import javax.sql.DataSource; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; +import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.annotation.Import; @@ -64,6 +66,8 @@ class JpaEventPublicationRepositoryIntegrationTests { private static final EventSerializer eventSerializer = mock(EventSerializer.class); + @Autowired EntityManager entityManager; + @Configuration @Import(JpaEventPublicationConfiguration.class) static class TestConfig { @@ -167,6 +171,34 @@ class JpaEventPublicationRepositoryIntegrationTests { assertThat(actual).isEmpty(); } + @Test // GH-20 + void shouldDeleteCompletedEvents() { + TestEvent testEvent1 = new TestEvent("abc"); + String serializedEvent1 = "{\"eventId\":\"abc\"}"; + TestEvent testEvent2 = new TestEvent("def"); + String serializedEvent2 = "{\"eventId\":\"def\"}"; + + when(eventSerializer.serialize(testEvent1)).thenReturn(serializedEvent1); + when(eventSerializer.deserialize(serializedEvent1, TestEvent.class)).thenReturn(testEvent1); + when(eventSerializer.serialize(testEvent2)).thenReturn(serializedEvent2); + when(eventSerializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); + + CompletableEventPublication publication1 = CompletableEventPublication.of(testEvent1, TARGET_IDENTIFIER); + CompletableEventPublication publication2 = CompletableEventPublication.of(testEvent2, TARGET_IDENTIFIER); + + repository.create(publication1); + repository.create(publication2); + + repository.update(publication1.markCompleted()); + + repository.deleteCompletedPublications(); + + var eventPublications = entityManager.createQuery( + "select p from JpaEventPublication p", JpaEventPublication.class).getResultList(); + assertThat(eventPublications).hasSize(1); + assertThat(eventPublications.get(0).getSerializedEvent()).isEqualTo(serializedEvent2); + } + @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/MongoDbEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java index bf76801d..35266a7b 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 @@ -102,6 +102,12 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0)); } + @Override + public void deleteCompletedPublications() { + var query = Query.query(Criteria.where("completionDate").ne(null)); + mongoTemplate.remove(query, MongoDbEventPublication.class); + } + private List findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( // Object event, PublicationTargetIdentifier targetIdentifier) { 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 81a04f73..f57e8fb7 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 @@ -28,6 +28,8 @@ 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.data.mongodb.core.MongoTemplate; +import org.springframework.data.mongodb.core.query.Criteria; +import org.springframework.data.mongodb.core.query.Query; import org.springframework.modulith.events.CompletableEventPublication; import org.springframework.modulith.events.EventPublication; import org.springframework.modulith.events.PublicationTargetIdentifier; @@ -163,6 +165,31 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb { } } + @Nested + class DeleteCompletedPublications { + + @Test // GH-20 + void shouldDeleteCompletedEvents() { + TestEvent testEvent1 = new TestEvent("abc"); + TestEvent testEvent2 = new TestEvent("def"); + + CompletableEventPublication publication1 = CompletableEventPublication.of(testEvent1, TARGET_IDENTIFIER); + CompletableEventPublication publication2 = CompletableEventPublication.of(testEvent2, TARGET_IDENTIFIER); + + repository.create(publication1); + repository.create(publication2); + + repository.update(publication1.markCompleted()); + + repository.deleteCompletedPublications(); + + var eventPublications = mongoTemplate.findAll(MongoDbEventPublication.class); + + assertThat(eventPublications).hasSize(1); + assertThat(eventPublications.get(0).getEvent()).isEqualTo(testEvent2); + } + } + @Value private static final class TestEvent { String eventId;