From 06f4cc70b027c1dbe2ab850312297b5fdce4306c Mon Sep 17 00:00:00 2001 From: Oliver Drotbohm Date: Tue, 6 Aug 2024 17:31:16 +0200 Subject: [PATCH] GH-258 - Optimize publication completion to issue by-id query. DefaultEventPublicationRegistry now tracks the event publications currently in progress, so that the completion step can use the database identifier to issue an update statement solely based on that. --- .../core/DefaultEventPublicationRegistry.java | 135 +++++++++++++++- .../events/core/EventPublicationRegistry.java | 25 +++ .../core/EventPublicationRepository.java | 11 +- .../events/core/TargetEventPublication.java | 23 ++- .../support/CompletionRegisteringAdvisor.java | 33 ++-- ...PersistentApplicationEventMulticaster.java | 38 +---- .../InMemoryEventPublicationRepository.java | 152 ++++++++++++++++++ ...ationEventMulticasterIntegrationTests.java | 22 +-- ...tApplicationEventMulticasterUnitTests.java | 3 +- .../jdbc/JdbcEventPublicationRepository.java | 19 +++ ...PublicationRepositoryIntegrationTests.java | 13 ++ .../jpa/JpaEventPublicationRepository.java | 19 +++ .../MongoDbEventPublicationRepository.java | 12 ++ ...MongoDbEventPublicationRepositoryTest.java | 13 ++ .../Neo4jEventPublicationRepository.java | 19 +++ .../Neo4jEventPublicationRepositoryTest.java | 13 ++ 16 files changed, 484 insertions(+), 66 deletions(-) create mode 100644 spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/InMemoryEventPublicationRepository.java 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 4fc12e51..a4004a0f 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 @@ -17,13 +17,19 @@ package org.springframework.modulith.events.core; import java.time.Clock; import java.time.Duration; +import java.time.Instant; import java.util.Collection; +import java.util.HashSet; +import java.util.Optional; +import java.util.Set; +import java.util.function.Consumer; import java.util.function.Predicate; import java.util.stream.Stream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; import org.springframework.beans.factory.DisposableBean; +import org.springframework.lang.Nullable; import org.springframework.modulith.events.CompletedEventPublications; import org.springframework.modulith.events.EventPublication; import org.springframework.transaction.annotation.Propagation; @@ -46,6 +52,7 @@ public class DefaultEventPublicationRegistry private final EventPublicationRepository events; private final Clock clock; + private final PublicationsInProgress inProgress; /** * Creates a new {@link DefaultEventPublicationRegistry} for the given {@link EventPublicationRepository}. @@ -60,6 +67,7 @@ public class DefaultEventPublicationRegistry this.events = events; this.clock = clock; + this.inProgress = new PublicationsInProgress(); } /* @@ -72,6 +80,7 @@ public class DefaultEventPublicationRegistry return listeners.map(it -> TargetEventPublication.of(event, it, clock.instant())) .peek(it -> LOGGER.debug(REGISTER, it.getEvent().getClass().getName(), it.getTargetIdentifier().getValue())) .map(events::create) + .map(inProgress::register) .toList(); } @@ -110,7 +119,24 @@ public class DefaultEventPublicationRegistry LOGGER.debug("Marking publication of event {} to listener {} completed.", // event.getClass().getName(), targetIdentifier.getValue()); - events.markCompleted(event, targetIdentifier, clock.instant()); + Instant now = clock.instant(); + Runnable fallback = () -> events.markCompleted(event, targetIdentifier, now); + Consumer optimized = it -> { + events.markCompleted(it.getIdentifier(), now); + inProgress.unregister(event, targetIdentifier); + }; + + inProgress.getPublication(event, targetIdentifier) + .ifPresentOrElse(optimized, fallback); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRegistry#markFailed(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier) + */ + @Override + public void markFailed(Object event, PublicationTargetIdentifier targetIdentifier) { + inProgress.unregister(event, targetIdentifier); } /* @@ -161,6 +187,42 @@ public class DefaultEventPublicationRegistry events.deleteCompletedPublicationsBefore(now.minus(duration)); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRegistry#processIncompletePublications(java.util.function.Predicate, java.util.function.Consumer, java.time.Duration) + */ + @Override + public void processIncompletePublications(Predicate filter, + Consumer consumer, @Nullable Duration duration) { + + var message = duration != null ? " older than %s".formatted(duration) : ""; + + LOGGER.debug("Looking up incomplete event publications {}… ", message); + + var publications = duration == null // + ? findIncompletePublications() // + : findIncompletePublicationsOlderThan(duration); + + LOGGER.debug(getConfirmationMessage(publications) + " found."); + + publications.stream() // + .filter(filter) // + .forEach(it -> { + + try { + + inProgress.register(it); + consumer.accept(it); + + } catch (Exception o_O) { + + if (LOGGER.isErrorEnabled()) { + LOGGER.error("Error republishing event publication " + it, o_O); + } + } + }); + } + /* * (non-Javadoc) * @see org.springframework.beans.factory.DisposableBean#destroy() @@ -186,4 +248,75 @@ public class DefaultEventPublicationRegistry LOGGER.info("{} {} - {}", prefix, it.getEvent().getClass().getName(), it.getTargetIdentifier().getValue()); } } + + private static String getConfirmationMessage(Collection publications) { + + var size = publications.size(); + + return switch (publications.size()) { + case 0 -> "No publication"; + case 1 -> "1 publication"; + default -> size + " publications"; + }; + } + + /** + * All {@link TargetEventPublication}s currently processed. + * + * @author Oliver Drotbohm + * @since 1.3 + */ + static class PublicationsInProgress { + + private final Set publications = new HashSet<>(); + + /** + * Registers the given {@link TargetEventPublication} as currently processed. + * + * @param publication must not be {@literal null}. + * @return will never be {@literal null}. + */ + TargetEventPublication register(TargetEventPublication publication) { + + Assert.notNull(publication, "TargetEventPublication must not be null!"); + + publications.add(publication); + + return publication; + } + + /** + * Unregisters the {@link TargetEventPublication} associated with the given event and + * {@link PublicationTargetIdentifier}. + * + * @param event must not be {@literal null}. + * @param identifier must not be {@literal null}. + */ + void unregister(Object event, PublicationTargetIdentifier identifier) { + + Assert.notNull(event, "Event must not be null!"); + Assert.notNull(identifier, "PublicationTargetIdentifier must not be null!"); + + getPublication(event, identifier) + .ifPresent(publications::remove); + } + + /** + * Returns the {@link TargetEventPublication} associated with the given event and + * {@link PublicationTargetIdentifier}. + * + * @param event must not be {@literal null}. + * @param identifier must not be {@literal null}. + * @return will never be {@literal null}. + */ + Optional getPublication(Object event, PublicationTargetIdentifier identifier) { + + Assert.notNull(event, "Event must not be null!"); + Assert.notNull(identifier, "PublicationTargetIdentifier must not be null!"); + + return publications.stream() + .filter(it -> it.isAssociatedWith(event, identifier)) + .findFirst(); + } + } } diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRegistry.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRegistry.java index 989c71ba..5b99151a 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRegistry.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRegistry.java @@ -17,9 +17,13 @@ package org.springframework.modulith.events.core; import java.time.Duration; import java.util.Collection; +import java.util.function.Consumer; +import java.util.function.Predicate; import java.util.stream.Stream; import org.springframework.context.ApplicationListener; +import org.springframework.lang.Nullable; +import org.springframework.modulith.events.EventPublication; /** * A registry to capture event publications to {@link ApplicationListener}s. Allows to register those publications, mark @@ -64,10 +68,31 @@ public interface EventPublicationRegistry { */ void markCompleted(Object event, PublicationTargetIdentifier targetIdentifier); + /** + * Marks the publication for the given event and {@link PublicationTargetIdentifier} as failed. + * + * @param event must not be {@literal null}. + * @param targetIdentifier must not be {@literal null}. + * @since 1.3 + */ + void markFailed(Object event, PublicationTargetIdentifier targetIdentifier); + /** * Deletes all completed {@link TargetEventPublication}s that have been completed before the given {@link Duration}. * * @param duration must not be {@literal null}. */ void deleteCompletedPublicationsOlderThan(Duration duration); + + /** + * Processes all incomplete event publications that have been published before the given duration in relation to "now" + * by applying the given filter passing all remaining instances to the given {@link Consumer}. + * + * @param filter must not be {@literal null}. + * @param consumer must not be {@literal null}. + * @param duration can be {@literal null}. + * @since 1.3 + */ + void processIncompletePublications(Predicate filter, + Consumer consumer, @Nullable Duration duration); } 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 2764ab03..cd827675 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 @@ -52,7 +52,7 @@ public interface EventPublicationRepository { publication.markCompleted(completionDate); - markCompleted(publication.getEvent(), publication.getTargetIdentifier(), completionDate); + markCompleted(publication.getIdentifier(), completionDate); } /** @@ -65,6 +65,15 @@ public interface EventPublicationRepository { */ void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate); + /** + * Marks the publication with the given identifier completed at the given {@link Instant}. + * + * @param identifier must not be {@literal null}. + * @param completionDate must not be {@literal null}. + * @since 1.3 + */ + void markCompleted(UUID identifier, Instant completionDate); + /** * Returns all {@link TargetEventPublication}s that have not been completed yet. * diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java index 3fe56acd..523e54d2 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java @@ -30,9 +30,9 @@ import org.springframework.util.Assert; public interface TargetEventPublication extends Completable, org.springframework.modulith.events.EventPublication { /** - * Creates a {@link TargetEventPublication} for the given event an listener identifier using a default {@link Instant}. - * Prefer using {@link #of(Object, PublicationTargetIdentifier, Instant)} with a dedicated {@link Instant} obtained - * from a {@link Clock}. + * Creates a {@link TargetEventPublication} for the given event an listener identifier using a default + * {@link Instant}. Prefer using {@link #of(Object, PublicationTargetIdentifier, Instant)} with a dedicated + * {@link Instant} obtained from a {@link Clock}. * * @param event must not be {@literal null}. * @param id must not be {@literal null}. @@ -66,7 +66,6 @@ public interface TargetEventPublication extends Completable, org.springframework * Returns whether the publication is identified by the given {@link PublicationTargetIdentifier}. * * @param identifier must not be {@literal null}. - * @return */ default boolean isIdentifiedBy(PublicationTargetIdentifier identifier) { @@ -74,4 +73,20 @@ public interface TargetEventPublication extends Completable, org.springframework return this.getTargetIdentifier().equals(identifier); } + + /** + * Returns whether the {@link TargetEventPublication} is associated with the given event and + * {@link PublicationTargetIdentifier}. + * + * @param event must not be {@literal null}. + * @param identifier must not be {@literal null}. + * @since 1.3 + */ + default boolean isAssociatedWith(Object event, PublicationTargetIdentifier identifier) { + + Assert.notNull(event, "Event must not be null!"); + Assert.notNull(identifier, "PublicationTargetIdentifier must not be null!"); + + return isIdentifiedBy(identifier) && getEvent().equals(event); + } } diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java index 66357631..7ca33d09 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/CompletionRegisteringAdvisor.java @@ -32,7 +32,6 @@ import org.springframework.aop.support.annotation.AnnotationMatchingPointcut; import org.springframework.core.Ordered; import org.springframework.core.annotation.AnnotatedElementUtils; import org.springframework.lang.NonNull; -import org.springframework.lang.Nullable; import org.springframework.modulith.events.core.EventPublicationRegistry; import org.springframework.modulith.events.core.PublicationTargetIdentifier; import org.springframework.transaction.event.TransactionPhase; @@ -139,9 +138,8 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor { static class CompletionRegisteringMethodInterceptor implements MethodInterceptor, Ordered { private static final Logger LOG = LoggerFactory.getLogger(CompletionRegisteringMethodInterceptor.class); - - private static final ConcurrentLruCache ADAPTERS = new ConcurrentLruCache<>( - 100, CompletionRegisteringMethodInterceptor::createAdapter); + private static final ConcurrentLruCache LISTENER_IDS = new ConcurrentLruCache<>( + 100, CompletionRegisteringMethodInterceptor::lookupListenerId); private final @NonNull Supplier registry; @@ -180,14 +178,14 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor { return it; }) .exceptionallyCompose(it -> { - handleFailure(method, it); + handleFailure(method, argument, it); return CompletableFuture.failedFuture(it); }); } } catch (Throwable o_O) { - handleFailure(method, o_O); + handleFailure(method, argument, o_O); throw o_O; } @@ -206,8 +204,9 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor { return Ordered.HIGHEST_PRECEDENCE + 10; } - @Nullable - private static Void handleFailure(Method method, Throwable o_O) { + private void handleFailure(Method method, Object event, Throwable o_O) { + + markFailed(method, event); if (LOG.isDebugEnabled()) { LOG.debug("Invocation of listener {} failed. Leaving event publication uncompleted.", method, o_O); @@ -215,20 +214,28 @@ public class CompletionRegisteringAdvisor extends AbstractPointcutAdvisor { LOG.info("Invocation of listener {} failed with message {}. Leaving event publication uncompleted.", method, o_O.getMessage()); } - - return null; } private void markCompleted(Method method, Object event) { // Mark publication complete if the method is a transactional event listener. - String adapterId = ADAPTERS.get(method).getListenerId(); + String adapterId = LISTENER_IDS.get(method); PublicationTargetIdentifier identifier = PublicationTargetIdentifier.of(adapterId); registry.get().markCompleted(event, identifier); } - private static TransactionalApplicationListenerMethodAdapter createAdapter(Method method) { - return new TransactionalApplicationListenerMethodAdapter(null, method.getDeclaringClass(), method); + private void markFailed(Method method, Object event) { + + // Mark publication complete if the method is a transactional event listener. + String adapterId = LISTENER_IDS.get(method); + PublicationTargetIdentifier identifier = PublicationTargetIdentifier.of(adapterId); + registry.get().markFailed(event, identifier); + } + + @SuppressWarnings("null") + private static String lookupListenerId(Method method) { + return new TransactionalApplicationListenerMethodAdapter(null, method.getDeclaringClass(), method) + .getListenerId(); } } } diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticaster.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticaster.java index 5172c5db..a0e315bb 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticaster.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticaster.java @@ -196,32 +196,7 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv private void doResubmitUncompletedPublicationsOlderThan(@Nullable Duration duration, Predicate filter) { - var message = duration != null ? " older than %s".formatted(duration) : ""; - var registry = this.registry.get(); - - LOGGER.debug("Looking up incomplete event publications {}… ", message); - - var publications = duration == null // - ? registry.findIncompletePublications() // - : registry.findIncompletePublicationsOlderThan(duration); - - LOGGER.debug(getConfirmationMessage(publications) + " found."); - - publications.stream() // - .filter(filter) // - .forEach(it -> { - - try { - - invokeTargetListener(it); - - } catch (Exception o_O) { - - if (LOGGER.isErrorEnabled()) { - LOGGER.error("Error republishing event publication " + it, o_O); - } - } - }); + registry.get().processIncompletePublications(filter, this::invokeTargetListener, duration); } private static ApplicationListener executeListenerWithCompletion(EventPublication publication, @@ -281,17 +256,6 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv : (boolean) ReflectionUtils.invokeMethod(LEGACY_SHOULD_HANDLE, candidate, event, new Object[] { payload }); } - private static String getConfirmationMessage(Collection publications) { - - var size = publications.size(); - - return switch (publications.size()) { - case 0 -> "No publication"; - case 1 -> "1 publication"; - default -> size + " publications"; - }; - } - /** * First-class collection to work with transactional event listeners, i.e. {@link ApplicationListener} instances that * implement {@link TransactionalApplicationListener}. diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/InMemoryEventPublicationRepository.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/InMemoryEventPublicationRepository.java new file mode 100644 index 00000000..ecc2d04b --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/InMemoryEventPublicationRepository.java @@ -0,0 +1,152 @@ +/* + * Copyright 2024 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.core; + +import java.time.Instant; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Iterator; +import java.util.List; +import java.util.Optional; +import java.util.UUID; +import java.util.function.Predicate; + +/** + * In-memory implementation of {@link EventPublicationRepository} for testing purposes. + * + * @author Oliver Drotbohm + */ +public class InMemoryEventPublicationRepository + implements EventPublicationRepository, Iterable { + + private static final Predicate IS_COMPLETED = it -> it.getCompletionDate() != null; + + private Collection publications = new ArrayList<>(); + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#create(org.springframework.modulith.events.core.TargetEventPublication) + */ + @Override + public TargetEventPublication create(TargetEventPublication publication) { + + if (!publications.contains(publication)) { + publications.add(publication); + } + + return publication; + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier, java.time.Instant) + */ + @Override + public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) { + + publications.stream() + .filter(it -> it.isAssociatedWith(event, identifier)) + .findFirst() + .ifPresent(it -> it.markCompleted(completionDate)); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant) + */ + @Override + public void markCompleted(UUID identifier, Instant completionDate) { + + publications.stream() + .filter(it -> it.getIdentifier().equals(identifier)) + .findFirst() + .ifPresent(it -> it.markCompleted(completionDate)); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublications() + */ + @Override + public List findIncompletePublications() { + + return publications.stream() + .filter(IS_COMPLETED.negate()) + .toList(); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsPublishedBefore(java.time.Instant) + */ + @Override + public List findIncompletePublicationsPublishedBefore(Instant instant) { + + return publications.stream() + .filter(IS_COMPLETED.negate()) + .filter(it -> it.getPublicationDate().isBefore(instant)) + .toList(); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier) + */ + @Override + public Optional findIncompletePublicationsByEventAndTargetIdentifier(Object event, + PublicationTargetIdentifier targetIdentifier) { + + return publications.stream() + .filter(it -> it.isAssociatedWith(event, targetIdentifier)) + .findFirst(); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) + */ + @Override + public void deletePublications(List identifiers) { + publications.removeIf(it -> identifiers.contains(it.getIdentifier())); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deleteCompletedPublications() + */ + @Override + public void deleteCompletedPublications() { + publications.removeIf(IS_COMPLETED); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deleteCompletedPublicationsBefore(java.time.Instant) + */ + @Override + public void deleteCompletedPublicationsBefore(Instant instant) { + publications.removeIf(IS_COMPLETED.and(it -> it.getPublicationDate().isBefore(instant))); + } + + /* + * (non-Javadoc) + * @see java.lang.Iterable#iterator() + */ + @Override + public Iterator iterator() { + return publications.iterator(); + } +} diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterIntegrationTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterIntegrationTests.java index 198efeae..37d2bd71 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterIntegrationTests.java +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterIntegrationTests.java @@ -15,8 +15,7 @@ */ package org.springframework.modulith.events.support; -import static org.mockito.ArgumentMatchers.*; -import static org.mockito.Mockito.*; +import static org.assertj.core.api.Assertions.*; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -26,8 +25,8 @@ import org.springframework.context.ApplicationEventPublisher; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.modulith.events.config.EnablePersistentDomainEvents; +import org.springframework.modulith.events.core.InMemoryEventPublicationRepository; import org.springframework.modulith.events.core.TargetEventPublication; -import org.springframework.modulith.events.core.EventPublicationRepository; import org.springframework.stereotype.Component; import org.springframework.test.context.junit.jupiter.SpringExtension; import org.springframework.transaction.annotation.EnableTransactionManagement; @@ -47,8 +46,8 @@ class PersistentApplicationEventMulticasterIntegrationTests { static class TestConfiguration { @Bean - EventPublicationRepository repository() { - return mock(EventPublicationRepository.class); + InMemoryEventPublicationRepository repository() { + return new InMemoryEventPublicationRepository(); } @Bean @@ -58,16 +57,21 @@ class PersistentApplicationEventMulticasterIntegrationTests { } @Autowired ApplicationEventPublisher publisher; - @Autowired EventPublicationRepository repository; + @Autowired InMemoryEventPublicationRepository repository; @Test // GH-186, GH-239 void doesNotPublishGenericEventsToListeners() throws Exception { publisher.publishEvent(new SomeGenericEvent<>()); - verify(repository, never()).create(any(TargetEventPublication.class)); - publisher.publishEvent(new SomeOtherEvent()); - verify(repository).create(any(TargetEventPublication.class)); + assertThat(repository).isEmpty(); + + var event = new SomeOtherEvent(); + publisher.publishEvent(event); + + assertThat(repository) + .extracting(TargetEventPublication::getEvent) + .containsExactly(event); } @Component diff --git a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterUnitTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterUnitTests.java index 2959954b..b59d7d7d 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterUnitTests.java @@ -16,6 +16,7 @@ package org.springframework.modulith.events.support; import static org.assertj.core.api.Assertions.*; +import static org.mockito.ArgumentMatchers.*; import static org.mockito.Mockito.*; import lombok.AllArgsConstructor; @@ -72,7 +73,7 @@ class PersistentApplicationEventMulticasterUnitTests { multicaster.afterSingletonsInstantiated(); - verify(registry).findIncompletePublications(); + verify(registry).processIncompletePublications(any(), any(), any()); } @Test // GH-277 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 1b58dac6..32459a71 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 @@ -90,6 +90,13 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean AND SERIALIZED_EVENT = ? """; + private static final String SQL_STATEMENT_UPDATE_BY_ID = """ + UPDATE %s + SET COMPLETION_DATE = ? + WHERE + ID = ? + """; + private static final String SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID = """ SELECT * FROM %s @@ -133,6 +140,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean sqlStatementFindUncompleted, sqlStatementFindUncompletedBefore, sqlStatementUpdateByEventAndListenerId, + sqlStatementUpdateById, sqlStatementFindByEventAndListenerId, sqlStatementDelete, sqlStatementDeleteUncompleted, @@ -167,6 +175,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean 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); + this.sqlStatementUpdateById = SQL_STATEMENT_UPDATE_BY_ID.formatted(table); this.sqlStatementFindByEventAndListenerId = SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID.formatted(table); this.sqlStatementDelete = SQL_STATEMENT_DELETE.formatted(table); this.sqlStatementDeleteUncompleted = SQL_STATEMENT_DELETE_UNCOMPLETED.formatted(table); @@ -217,6 +226,16 @@ class JdbcEventPublicationRepository implements EventPublicationRepository, Bean serializer.serialize(event)); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant) + */ + @Override + @Transactional + public void markCompleted(UUID identifier, Instant completionDate) { + operations.update(sqlStatementUpdateById, Timestamp.from(completionDate), uuidToDatabase(identifier)); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier) 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 e441c6b6..d414bbcb 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 @@ -310,6 +310,19 @@ class JdbcEventPublicationRepositoryIntegrationTests { .isEqualTo(event); } + @Test // GH-258 + void marksPublicationAsCompletedById() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication.getIdentifier(), Instant.now()); + + assertThat(repository.findCompletedPublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(publication.getIdentifier()); + } + abstract String table(); private TargetEventPublication createPublication(Object event) { 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 d773d160..e9e527dc 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 @@ -85,6 +85,12 @@ class JpaEventPublicationRepository implements EventPublicationRepository { and p.listenerId = ?2 """; + private static final String MARK_COMPLETED_BY_ID = """ + update JpaEventPublication p + set p.completionDate = ?2 + where p.id = ?1 + """; + private static final String DELETE = """ delete from JpaEventPublication p @@ -153,6 +159,19 @@ class JpaEventPublicationRepository implements EventPublicationRepository { .executeUpdate(); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant) + */ + @Override + public void markCompleted(UUID identifier, Instant completionDate) { + + entityManager.createQuery(MARK_COMPLETED_BY_ID) + .setParameter(1, identifier) + .setParameter(2, completionDate) + .executeUpdate(); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.EventPublicationRepository#findIncompletePublications() 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 609bb469..0620f5ff 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 @@ -92,6 +92,18 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { mongoTemplate.findAndModify(byEventAndListenerId(event, identifier), update, MongoDbEventPublication.class); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant) + */ + @Override + public void markCompleted(UUID identifier, Instant completionDate) { + + var update = Update.update(COMPLETION_DATE, completionDate); + + mongoTemplate.findAndModify(query(where(ID).is(identifier)), update, MongoDbEventPublication.class); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublications() 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 709e2934..4022ea2f 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 @@ -144,6 +144,19 @@ class MongoDbEventPublicationRepositoryTest { .isEqualTo(event); } + @Test // GH-258 + void marksPublicationAsCompletedById() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication.getIdentifier(), Instant.now()); + + assertThat(repository.findCompletedPublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(publication.getIdentifier()); + } + 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 3c25763b..698933ca 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 @@ -110,6 +110,11 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .set(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).to(Cypher.parameter(COMPLETION_DATE))) .build(); + private static final Statement COMPLETE_BY_ID_STATEMENT = Cypher.match(EVENT_PUBLICATION_NODE) + .where(EVENT_PUBLICATION_NODE.property(ID).eq(Cypher.parameter(ID))) + .set(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).to(Cypher.parameter(COMPLETION_DATE))) + .build(); + private static final ResultStatement INCOMPLETE_STATEMENT = Cypher.match(EVENT_PUBLICATION_NODE) .where(EVENT_PUBLICATION_NODE.property(COMPLETION_DATE).isNull()) .returning(EVENT_PUBLICATION_NODE) @@ -185,6 +190,20 @@ class Neo4jEventPublicationRepository implements EventPublicationRepository { .run(); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#markCompleted(java.util.UUID, java.time.Instant) + */ + @Override + @Transactional + public void markCompleted(UUID identifier, Instant completionDate) { + + neo4jClient.query(renderer.render(COMPLETE_BY_ID_STATEMENT)) + .bind(Values.value(identifier.toString())).to(ID) + .bind(Values.value(completionDate.atOffset(ZoneOffset.UTC))).to(COMPLETION_DATE) + .run(); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier) 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 54472942..eec41bb3 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 @@ -235,6 +235,19 @@ class Neo4jEventPublicationRepositoryTest { .isEqualTo(event); } + @Test // GH-258 + void marksPublicationAsCompletedById() { + + var event = new TestEvent("first"); + var publication = createPublication(event); + + repository.markCompleted(publication.getIdentifier(), Instant.now()); + + assertThat(repository.findCompletedPublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(publication.getIdentifier()); + } + private TargetEventPublication createPublication(Object event) { var token = event.toString();