From 97a50519302e85573dd52e9c109277f0badd8bdc Mon Sep 17 00:00:00 2001 From: Dmitry Belyaev Date: Wed, 27 Jul 2022 10:53:16 +0200 Subject: [PATCH] GH-25 - Fix #findByEventAndTargetIdentifier in all repositories to return only incomplete EventPublications. Rename the method accordingly. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Signed-off-by: Björn Kieling --- .../DefaultEventPublicationRegistry.java | 2 +- .../events/EventPublicationRepository.java | 5 +-- .../MapEventPublicationRepository.java | 4 +-- .../jdbc/JdbcEventPublicationRepository.java | 9 +++-- ...PublicationRepositoryIntegrationTests.java | 33 ++++++++++++++----- .../jpa/JpaEventPublicationRepository.java | 32 +++++++++++++----- ...PublicationRepositoryIntegrationTests.java | 33 ++++++++++++++++++- .../MongoDbEventPublicationRepository.java | 12 ++++--- ...MongoDbEventPublicationRepositoryTest.java | 31 ++++++++++++----- 9 files changed, 122 insertions(+), 39 deletions(-) diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/DefaultEventPublicationRegistry.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/DefaultEventPublicationRegistry.java index af40ae20..45204609 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/DefaultEventPublicationRegistry.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/DefaultEventPublicationRegistry.java @@ -61,7 +61,7 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub Assert.notNull(event, "Domain event must not be null!"); Assert.notNull(targetIdentifier, "Listener identifier must not be null!"); - events.findByEventAndTargetIdentifier(event, targetIdentifier) // + events.findIncompletePublicationsByEventAndTargetIdentifier(event, targetIdentifier) // .map(DefaultEventPublicationRegistry::logCompleted) // .map(e -> CompletableEventPublication.of(e.getEvent(), e.getTargetIdentifier())) .ifPresent(it -> events.update(it.markCompleted())); 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 7860b07f..f214fcbd 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 @@ -51,11 +51,12 @@ public interface EventPublicationRepository { List findIncompletePublications(); /** - * Return the {@link EventPublication} for the given serialized event and listener identifier. + * Return the incomplete {@link EventPublication} for the given serialized event and listener identifier. * * @param event must not be {@literal null}. * @param targetIdentifier must not be {@literal null}. * @return will never be {@literal null}. */ - Optional findByEventAndTargetIdentifier(Object event, PublicationTargetIdentifier targetIdentifier); + Optional findIncompletePublicationsByEventAndTargetIdentifier( // + Object event, PublicationTargetIdentifier targetIdentifier); } diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/MapEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/MapEventPublicationRepository.java index 90321fb9..17ca06ed 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/MapEventPublicationRepository.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/MapEventPublicationRepository.java @@ -69,8 +69,8 @@ public class MapEventPublicationRepository implements EventPublicationRepository } @Override - public Optional findByEventAndTargetIdentifier(Object event, - PublicationTargetIdentifier targetIdentifier) { + public Optional findIncompletePublicationsByEventAndTargetIdentifier( + Object event, PublicationTargetIdentifier targetIdentifier) { return Optional.ofNullable(events.get(new Key(event, targetIdentifier))); } 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 08adbe21..d7b3dada 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 @@ -67,7 +67,10 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { """; private static final String SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID = """ SELECT * FROM EVENT_PUBLICATION - WHERE SERIALIZED_EVENT = ? AND LISTENER_ID = ? + WHERE + SERIALIZED_EVENT = ? + AND LISTENER_ID = ? + AND COMPLETION_DATE IS NULL ORDER BY PUBLICATION_DATE """; @@ -109,8 +112,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { @Override @Transactional(readOnly = true) - public Optional findByEventAndTargetIdentifier(Object event, - PublicationTargetIdentifier targetIdentifier) { + 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()); 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 b20c0c78..6072b9e3 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 @@ -86,7 +86,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { assertThat(it.getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier()); }); - assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)) + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)) .isPresent(); // Complete publication @@ -130,8 +130,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { @Nested class FindByEventAndTargetIdentifier { - @Test - // GH-3 + @Test // GH-3 void shouldTolerateEmptyResult() { var testEvent = new TestEvent("id"); @@ -139,11 +138,29 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent)).thenReturn(serializedEvent); - assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); } - @Test - // GH-3 + @Test // GH-3 + void shouldNotReturnCompletedEvents() { + + var testEvent = new TestEvent("id1"); + var serializedEvent = "{\"eventId\":\"id1\"}"; + + when(serializer.serialize(testEvent)).thenReturn(serializedEvent); + when(serializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); + + var publication = CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER); + + repository.create(publication); + repository.update(publication.markCompleted()); + + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + + assertThat(actual).isEmpty(); + } + + @Test // GH-3 // void shouldReturnTheOldestEvent() throws Exception { var testEvent = new TestEvent("id"); @@ -159,7 +176,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { repository.create(publicationNew); repository.create(publicationOld); - var actual = repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); assertThat(actual).hasValueSatisfying(it -> { assertThat(it.getPublicationDate()) // @@ -182,7 +199,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { operations.update("UPDATE EVENT_PUBLICATION SET EVENT_TYPE='abc'"); - assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); } } } 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 0b7602d4..ddc30201 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 @@ -40,8 +40,21 @@ 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 where p.serializedEvent = ?1 and p.listenerId = ?2"; - private static String INCOMPLETE = "select p from JpaEventPublication p where p.completionDate is null"; + private static String BY_EVENT_AND_LISTENER_ID = """ + select p + from JpaEventPublication p + where + p.serializedEvent = ?1 + and p.listenerId = ?2 + and p.completionDate is null + """; + + private static String INCOMPLETE = """ + select p + from JpaEventPublication p + where + p.completionDate is null + """; private final EntityManager entityManager; private final EventSerializer serializer; @@ -62,8 +75,8 @@ public class JpaEventPublicationRepository implements EventPublicationRepository var id = publication.getTargetIdentifier().getValue(); var event = publication.getEvent(); - findEntityBySerializedEventAndListenerId(event, id) // - .setCompletionDate(publication.getCompletionDate().orElse(null)); + findEntityBySerializedEventAndListenerIdAndCompletionDateNull(event, id) // + .ifPresent(entity -> entity.setCompletionDate(publication.getCompletionDate().orElse(null))); return publication; } @@ -80,14 +93,15 @@ public class JpaEventPublicationRepository implements EventPublicationRepository @Override @Transactional(readOnly = true) - public Optional findByEventAndTargetIdentifier(Object event, - PublicationTargetIdentifier targetIdentifier) { + public Optional findIncompletePublicationsByEventAndTargetIdentifier( // + Object event, PublicationTargetIdentifier targetIdentifier) { - return Optional.ofNullable(findEntityBySerializedEventAndListenerId(event, targetIdentifier.getValue())) + return findEntityBySerializedEventAndListenerIdAndCompletionDateNull(event, targetIdentifier.getValue()) .map(this::entityToDomain); } - private JpaEventPublication findEntityBySerializedEventAndListenerId(Object event, String listenerId) { + private Optional findEntityBySerializedEventAndListenerIdAndCompletionDateNull( // + Object event, String listenerId) { var serializedEvent = serializeEvent(event); @@ -95,7 +109,7 @@ public class JpaEventPublicationRepository implements EventPublicationRepository .setParameter(1, serializedEvent) .setParameter(2, listenerId); - return query.getSingleResult(); + return query.getResultStream().findFirst(); } private String serializeEvent(Object event) { 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 8be138d5..984274a7 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 @@ -127,7 +127,7 @@ class JpaEventPublicationRepositoryIntegrationTests { assertThat(eventPublications).hasSize(1); assertThat(eventPublications.get(0).getEvent()).isEqualTo(publication.getEvent()); assertThat(eventPublications.get(0).getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier()); - assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent(); + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent(); // Complete publication repository.update(publication.markCompleted()); @@ -135,6 +135,37 @@ class JpaEventPublicationRepositoryIntegrationTests { assertThat(repository.findIncompletePublications()).isEmpty(); } + @Test + void shouldTolerateEmptyResult() { + + var testEvent = new TestEvent("id"); + var serializedEvent = "{\"eventId\":\"id\"}"; + + when(eventSerializer.serialize(testEvent)).thenReturn(serializedEvent); + + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); + } + + @Test + void shouldNotReturnCompletedEvents() { + + TestEvent testEvent = new TestEvent("abc"); + String serializedEvent = "{\"eventId\":\"abc\"}"; + + when(eventSerializer.serialize(testEvent)).thenReturn(serializedEvent); + when(eventSerializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); + + CompletableEventPublication publication = CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER); + + // Store publication + repository.create(publication); + repository.update(publication.markCompleted()); + + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + + assertThat(actual).isEmpty(); + } + @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 7b5efb34..bf76801d 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 @@ -68,7 +68,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { @Override public EventPublication update(CompletableEventPublication publication) { - return findDocumentsByEventAndTargetIdentifier(publication.getEvent(), publication.getTargetIdentifier()) + return findDocumentsByEventAndTargetIdentifierAndCompletionDateNull(publication.getEvent(), publication.getTargetIdentifier()) .stream() .findFirst() .map(document -> document.setCompletionDate(publication.getCompletionDate().orElse(null))) @@ -88,10 +88,11 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { } @Override - public Optional findByEventAndTargetIdentifier( + public Optional findIncompletePublicationsByEventAndTargetIdentifier( Object event, PublicationTargetIdentifier targetIdentifier) { - var documents = findDocumentsByEventAndTargetIdentifier(event, targetIdentifier); + var documents = findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( // + event, targetIdentifier); var results = documents .stream() // .map(this::documentToDomain) // @@ -101,7 +102,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0)); } - private List findDocumentsByEventAndTargetIdentifier( + private List findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( // Object event, PublicationTargetIdentifier targetIdentifier) { // we need to enforce writing of the type information @@ -109,7 +110,8 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { var query = Query // .query(Criteria // .where("event").is(eventAsMongoType) // - .and("listenerId").is(targetIdentifier.getValue())) // + .and("listenerId").is(targetIdentifier.getValue()) // + .and("completionDate").isNull()) // .with(Sort.by("publicationDate").ascending()); return mongoTemplate.find(query, 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 0f143e11..81a04f73 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 @@ -74,7 +74,7 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb { assertThat(eventPublications.get(0).getEvent()).isEqualTo(publication.getEvent()); assertThat(eventPublications.get(0).getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier()); - assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent(); + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent(); // Complete publication repository.update(publication.markCompleted()); @@ -113,7 +113,7 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb { repository.create(CompletableEventPublication.of( testEvent1, PublicationTargetIdentifier.of(TARGET_IDENTIFIER.getValue() + "!"))); - var actual = repository.findByEventAndTargetIdentifier(testEvent1, TARGET_IDENTIFIER); + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent1, TARGET_IDENTIFIER); assertThat(actual).hasValueSatisfying(it -> { assertThat(it.getEvent()).isEqualTo(testEvent1); @@ -126,7 +126,23 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb { var testEvent = new TestEvent("id"); - assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty(); + } + + @Test + void shouldNotReturnCompletedEvents() { + + TestEvent testEvent = new TestEvent("abc"); + + CompletableEventPublication publication = CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER); + + // Store publication + repository.create(publication); + repository.update(publication.markCompleted()); + + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + + assertThat(actual).isEmpty(); } @Test // GH-4 @@ -139,12 +155,11 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb { Thread.sleep(10); repository.create(CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER)); - var actual = repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); - assertThat(actual).hasValueSatisfying(it -> { - assertThat(it.getPublicationDate()) // - .isCloseTo(publicationOld.getPublicationDate(), within(1, ChronoUnit.MILLIS)); - }); + assertThat(actual).hasValueSatisfying(it -> // + assertThat(it.getPublicationDate()) // + .isCloseTo(publicationOld.getPublicationDate(), within(1, ChronoUnit.MILLIS))); } }