From 6afc00c554baeb6aabc556c092b5e124be5e6ef0 Mon Sep 17 00:00:00 2001 From: Oliver Drotbohm Date: Tue, 16 Jan 2024 17:44:49 +0100 Subject: [PATCH] GH-452 - Introduce EventPublicationRepository.findCompletedPublications(). The default EventPublicationRegistry previously erroneously called find*In*completePublications(), i.e. returned incomplete publications when it was supposed to look up the completed ones. To be able to do the latter we need to extend EventPublicationRepository and introduce a query execution for all completed publications. As we'd normally only chance API in major versions, but the bugfix needing to go into a bugfix release, the new method is introduced as default method rejecting the execution in case a currently existing repository implementation does not implement it. This allows us to adapt the implementations we ship to support the bugfix, but will require other implementations to ship an adapted version. This means we can ship a release that's not breaking assuming one of the official store implementations is used. --- .../core/DefaultEventPublicationRegistry.java | 3 +-- .../core/EventPublicationRepository.java | 11 ++++++++ .../jdbc/JdbcEventPublicationRepository.java | 21 ++++++++++++++-- ...PublicationRepositoryIntegrationTests.java | 15 +++++++++++ .../jpa/JpaEventPublicationRepository.java | 23 +++++++++++++++++ ...PublicationRepositoryIntegrationTests.java | 15 +++++++++++ .../MongoDbEventPublicationRepository.java | 9 +++++++ ...MongoDbEventPublicationRepositoryTest.java | 15 +++++++++++ .../Neo4jEventPublicationRepository.java | 21 ++++++++++++++++ .../Neo4jEventPublicationRepositoryTest.java | 25 +++++++++++++++++++ 10 files changed, 154 insertions(+), 4 deletions(-) diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java index 59eb92be..436d8799 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java @@ -24,7 +24,6 @@ import java.util.stream.Stream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.DisposableBean; -import org.springframework.context.ApplicationListener; import org.springframework.modulith.events.CompletedEventPublications; import org.springframework.modulith.events.EventPublication; import org.springframework.transaction.annotation.Propagation; @@ -132,7 +131,7 @@ public class DefaultEventPublicationRegistry */ @Override public Collection findAll() { - return findIncompletePublications(); + return events.findCompletedPublications(); } /* diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java index a1eaa71f..456889e1 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java @@ -92,6 +92,17 @@ public interface EventPublicationRepository { Optional findIncompletePublicationsByEventAndTargetIdentifier( // Object event, PublicationTargetIdentifier targetIdentifier); + /** + * Returns all completed event publications currently found in the system. + * + * @return will never be {@literal null}. + * @since 1.1.2 + */ + default List findCompletedPublications() { + throw new UnsupportedOperationException( + "Your store implementation does not support looking up completed publications!"); + } + /** * Deletes all publications with the given identifiers. * 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 73fdca93..58bb8450 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 @@ -30,8 +30,6 @@ import java.util.stream.IntStream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.jdbc.core.JdbcOperations; -import org.springframework.jdbc.core.ResultSetExtractor; -import org.springframework.jdbc.core.RowMapper; import org.springframework.lang.Nullable; import org.springframework.modulith.events.core.EventPublicationRepository; import org.springframework.modulith.events.core.EventSerializer; @@ -56,6 +54,13 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { VALUES (?, ?, ?, ?, ?) """; + private static final String SQL_STATEMENT_FIND_COMPLETED = """ + SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT + FROM EVENT_PUBLICATION + WHERE COMPLETION_DATE IS NOT NULL + ORDER BY PUBLICATION_DATE ASC + """; + private static final String SQL_STATEMENT_FIND_UNCOMPLETED = """ SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT FROM EVENT_PUBLICATION @@ -189,6 +194,18 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { return result == null ? Optional.empty() : result.stream().findFirst(); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findCompletedPublications() + */ + @Override + public List findCompletedPublications() { + + var result = operations.query(SQL_STATEMENT_FIND_COMPLETED, this::resultSetToPublications); + + return result == null ? Collections.emptyList() : result; + } + @Override @Transactional(readOnly = true) @SuppressWarnings("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 f63ea4fd..e82cd5ad 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 @@ -294,6 +294,21 @@ class JdbcEventPublicationRepositoryIntegrationTests { .element(0).extracting(TargetEventPublication::getIdentifier).isEqualTo(first.getIdentifier()); } + @Test // GH-452 + void findsCompletedPublications() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication, Instant.now()); + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } + private TargetEventPublication createPublication(Object event) { var token = event.toString(); 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 4ecc417a..ef5a26ee 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 @@ -50,6 +50,15 @@ class JpaEventPublicationRepository implements EventPublicationRepository { and p.completionDate is null """; + private static String COMPLETE = """ + select p + from JpaEventPublication p + where + p.completionDate is not null + order by + p.publicationDate asc + """; + private static String INCOMPLETE = """ select p from JpaEventPublication p @@ -186,6 +195,20 @@ class JpaEventPublicationRepository implements EventPublicationRepository { .map(this::entityToDomain); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findCompletedPublications() + */ + @Override + public List findCompletedPublications() { + + return entityManager.createQuery(COMPLETE, JpaEventPublication.class) + .getResultList() + .stream() + .map(this::entityToDomain) + .toList(); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) 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 fad9d238..946443d9 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 @@ -270,6 +270,21 @@ class JpaEventPublicationRepositoryIntegrationTests { .element(0).extracting(TargetEventPublication::getIdentifier).isEqualTo(first.getIdentifier()); } + @Test // GH-452 + void findsCompletedPublications() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication, Instant.now()); + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } + private TargetEventPublication createPublication(Object event) { var token = event.toString(); 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 a3339dab..92d95f4c 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 @@ -127,6 +127,15 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0)); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findCompletedPublications() + */ + @Override + public List findCompletedPublications() { + return readMapped(defaultQuery(where(COMPLETION_DATE).ne(null))); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) 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 f7c50b6a..3f0d443d 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 @@ -129,6 +129,21 @@ class MongoDbEventPublicationRepositoryTest { .element(0).extracting(TargetEventPublication::getIdentifier).isEqualTo(first.getIdentifier()); } + @Test // GH-452 + void findsCompletedPublications() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication, Instant.now()); + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } + private TargetEventPublication createPublication(Object event) { return createPublication(event, TARGET_IDENTIFIER); } 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 5849cb4c..99127712 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 @@ -17,6 +17,7 @@ package org.springframework.modulith.events.neo4j; import java.time.Instant; import java.time.ZoneOffset; +import java.util.ArrayList; import java.util.List; import java.util.Map; import java.util.Objects; @@ -45,6 +46,7 @@ import org.springframework.util.DigestUtils; * A {@link Neo4jClient} based implementation of {@link EventPublicationRepository}. * * @author Gerrit Meier + * @author Oliver Drotbohm * @since 1.1 */ @Transactional @@ -114,6 +116,12 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .orderBy(EVENT_PUBLICATION_NODE.property(PUBLICATION_DATE)) .build(); + private static final ResultStatement ALL_COMPLETED_STATEMENT = Cypher.match(EVENT_PUBLICATION_NODE) + .where(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).isNotNull()) + .returning(EVENT_PUBLICATION_NODE) + .orderBy(EVENT_PUBLICATION_NODE.property(PUBLICATION_DATE)) + .build(); + private final Neo4jClient neo4jClient; private final Renderer renderer; private final EventSerializer eventSerializer; @@ -225,6 +233,19 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .one(); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findCompletedPublications() + */ + @Override + public List findCompletedPublications() { + + return new ArrayList<>(neo4jClient.query(renderer.render(ALL_COMPLETED_STATEMENT)) + .fetchAs(TargetEventPublication.class) + .mappedBy(this::mapRecordToPublication) + .all()); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) 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 b767ff6d..127481bd 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 @@ -220,6 +220,31 @@ class Neo4jEventPublicationRepositoryTest { } } + @Test // GH-452 + void findsCompletedPublications() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication, Instant.now()); + + assertThat(repository.findCompletedPublications()) + .hasSize(1) + .element(0) + .extracting(TargetEventPublication::getEvent) + .isEqualTo(event); + } + + private TargetEventPublication createPublication(Object event) { + + var token = event.toString(); + + doReturn(token).when(eventSerializer).serialize(event); + doReturn(event).when(eventSerializer).deserialize(token, event.getClass()); + + return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER)); + } + @Value static class TestEvent { String eventId;