diff --git a/spring-modulith-events/pom.xml b/spring-modulith-events/pom.xml index 9c5162fc..d6278f71 100644 --- a/spring-modulith-events/pom.xml +++ b/spring-modulith-events/pom.xml @@ -14,6 +14,7 @@ Spring Modulith - Events + spring-modulith-events-api spring-modulith-events-core spring-modulith-events-jpa spring-modulith-events-jdbc diff --git a/spring-modulith-events/spring-modulith-events-api/pom.xml b/spring-modulith-events/spring-modulith-events-api/pom.xml new file mode 100644 index 00000000..f30568fa --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-api/pom.xml @@ -0,0 +1,35 @@ + + + 4.0.0 + + + org.springframework.modulith + spring-modulith-events + 1.1.0-SNAPSHOT + + + Spring Modulith - Events - API + spring-modulith-events-api + + + org.springframework.modulith.events.api + + + + + + org.springframework.modulith + spring-modulith-api + ${project.version} + + + + org.springframework + spring-context + + + + + diff --git a/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/CompletedEventPublications.java b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/CompletedEventPublications.java new file mode 100644 index 00000000..fb4bd82b --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/CompletedEventPublications.java @@ -0,0 +1,51 @@ +/* + * Copyright 2023 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; + +import java.time.Duration; +import java.util.Collection; +import java.util.function.Predicate; + +/** + * All {@link EventPublication}s that have already been completed. + * + * @author Oliver Drotbohm + * @since 1.1 + */ +public interface CompletedEventPublications { + + /** + * Returns all {@link EventPublication}s that have already been completed. + * + * @return will never be {@literal null}. + */ + Collection findAll(); + + /** + * Deletes all {@link EventPublication}s matching the given {@link Predicate}. Note that implementations will iterate + * all completed {@link EventPublication}s and apply the predicate in memory. + * + * @param filter must not be {@literal null}. + */ + void deletePublications(Predicate filter); + + /** + * Deletes all {@link EventPublication}s whose completion date is older than the given {@link Duration}. + * + * @param duration must not be {@literal null}. + */ + void deletePublicationsOlderThan(Duration duration); +} diff --git a/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventPublication.java b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventPublication.java new file mode 100644 index 00000000..dda8118b --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventPublication.java @@ -0,0 +1,92 @@ +/* + * Copyright 2023 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; + +import java.time.Instant; +import java.util.Optional; +import java.util.UUID; + +import org.springframework.context.ApplicationEvent; +import org.springframework.context.PayloadApplicationEvent; + +/** + * An event publication. + * + * @author Oliver Drotbohm + * @since 1.1 + */ +public interface EventPublication { + + /** + * Returns a unique identifier for this publication. + * + * @return will never be {@literal null}. + */ + UUID getIdentifier(); + + /** + * Returns the event that is published. + * + * @return + */ + Object getEvent(); + + /** + * Returns the event as Spring {@link ApplicationEvent}, effectively wrapping it into a + * {@link PayloadApplicationEvent} in case it's not one already. + * + * @return + */ + default ApplicationEvent getApplicationEvent() { + + Object event = getEvent(); + + return PayloadApplicationEvent.class.isInstance(event) // + ? PayloadApplicationEvent.class.cast(event) + : new PayloadApplicationEvent<>(this, event); + } + + /** + * Returns the time the event is published at. + * + * @return + */ + Instant getPublicationDate(); + + /** + * Returns the completion date of the publication. + * + * @return will never be {@literal null}. + */ + Optional getCompletionDate(); + + /** + * Returns whether the publication of the event has completed. + * + * @return will never be {@literal null}. + */ + default boolean isPublicationCompleted() { + return getCompletionDate().isPresent(); + } + + /* + * (non-Javadoc) + * @see java.lang.Comparable#compareTo(java.lang.Object) + */ + default int compareTo(EventPublication that) { + return this.getPublicationDate().compareTo(that.getPublicationDate()); + } +} diff --git a/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/IncompleteEventPublications.java b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/IncompleteEventPublications.java new file mode 100644 index 00000000..f2b86404 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/IncompleteEventPublications.java @@ -0,0 +1,44 @@ +/* + * Copyright 2023 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; + +import java.time.Duration; +import java.util.function.Predicate; + +/** + * All uncompleted event publications. + * + * @author Oliver Drotbohm + * @since 1.1 + */ +public interface IncompleteEventPublications { + + /** + * Triggers the re-submission of events for which incomplete {@link EventPublication}s are registered. Note, that this + * will materialize all incomplete event publications. + * + * @param filter a {@link Predicate} to select the event publications for which to resubmit events. + */ + void resubmitIncompletePublications(Predicate filter); + + /** + * Triggers the re-submission of events for which incomplete {@link EventPublication}s are registered that exceed a + * certain age regarding their original publication date. + * + * @param duration must not be {@literal null}. + */ + void resubmitIncompletePublicationsOlderThan(Duration duration); +} diff --git a/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/package-info.java b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/package-info.java new file mode 100644 index 00000000..0f3c9311 --- /dev/null +++ b/spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/package-info.java @@ -0,0 +1,5 @@ +/** + * API of the event publication registry abstraction. + */ +@org.springframework.lang.NonNullApi +package org.springframework.modulith.events; diff --git a/spring-modulith-events/spring-modulith-events-core/pom.xml b/spring-modulith-events/spring-modulith-events-core/pom.xml index d1a84588..d4fadacb 100644 --- a/spring-modulith-events/spring-modulith-events-core/pom.xml +++ b/spring-modulith-events/spring-modulith-events-core/pom.xml @@ -18,6 +18,12 @@ + + org.springframework.modulith + spring-modulith-events-api + ${project.version} + + org.springframework spring-context diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/Completable.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/Completable.java index a8abe778..0a00da3a 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/Completable.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/Completable.java @@ -18,7 +18,7 @@ package org.springframework.modulith.events.core; import java.time.Instant; /** - * Internal interface to be able to mark {@link EventPublication} instances as completed. + * Internal interface to be able to mark {@link TargetEventPublication} instances as completed. * * @author Oliver Drotbohm */ diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublication.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublication.java index 668f4c18..d87988e0 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublication.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublication.java @@ -20,6 +20,7 @@ import java.util.Objects; import java.util.Optional; import java.util.UUID; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -27,7 +28,7 @@ import org.springframework.util.Assert; * * @author Oliver Drotbohm */ -class DefaultEventPublication implements EventPublication { +class DefaultEventPublication implements TargetEventPublication { private final UUID identifier; private final Object event; @@ -124,7 +125,7 @@ class DefaultEventPublication implements EventPublication { * @see java.lang.Object#equals(java.lang.Object) */ @Override - public boolean equals(Object obj) { + public boolean equals(@Nullable Object obj) { if (this == obj) { return true; 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 12b9c65b..59eb92be 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 @@ -18,13 +18,15 @@ package org.springframework.modulith.events.core; import java.time.Clock; import java.time.Duration; import java.util.Collection; -import java.util.List; +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.context.ApplicationListener; +import org.springframework.modulith.events.CompletedEventPublications; +import org.springframework.modulith.events.EventPublication; import org.springframework.transaction.annotation.Propagation; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; @@ -37,7 +39,8 @@ import org.springframework.util.Assert; * @author Björn Kieling * @author Dmitry Belyaev */ -public class DefaultEventPublicationRegistry implements DisposableBean, EventPublicationRegistry { +public class DefaultEventPublicationRegistry + implements DisposableBean, EventPublicationRegistry, CompletedEventPublications { private static final Logger LOGGER = LoggerFactory.getLogger(DefaultEventPublicationRegistry.class); private static final String REGISTER = "Registering publication of {} for {}."; @@ -65,9 +68,9 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub * @see org.springframework.modulith.events.EventPublicationRegistry#store(java.lang.Object, java.util.stream.Stream) */ @Override - public Collection store(Object event, Stream listeners) { + public Collection store(Object event, Stream listeners) { - return listeners.map(it -> EventPublication.of(event, it, clock.instant())) + 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) .toList(); @@ -78,10 +81,22 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub * @see org.springframework.modulith.events.EventPublicationRegistry#findIncompletePublications() */ @Override - public Collection findIncompletePublications() { + public Collection findIncompletePublications() { return events.findIncompletePublications(); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRegistry#findIncompletePublicationsOlderThan(java.time.Duration) + */ + @Override + public Collection findIncompletePublicationsOlderThan(Duration duration) { + + var reference = clock.instant().minus(duration); + + return events.findIncompletePublicationsPublishedBefore(reference); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.EventPublicationRegistry#markCompleted(java.lang.Object, org.springframework.modulith.events.PublicationTargetIdentifier) @@ -111,6 +126,44 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub events.deleteCompletedPublicationsBefore(clock.instant().minus(duration)); }; + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.CompletedEventPublications#findAll() + */ + @Override + public Collection findAll() { + return findIncompletePublications(); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.CompletedEventPublications#deletePublications(java.util.function.Predicate) + */ + @Override + public void deletePublications(Predicate filter) { + + var identifiers = findIncompletePublications().stream() + .filter(filter) + .map(TargetEventPublication::getIdentifier) + .toList(); + + events.deletePublications(identifiers); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.CompletedEventPublications#deletePublicationsOlderThan(java.time.Duration) + */ + @Override + public void deletePublicationsOlderThan(Duration duration) { + + var now = clock.instant(); + + deletePublications(event -> event.getCompletionDate() + .filter(date -> date.isBefore(now.minus(duration))) + .isPresent()); + } + /* * (non-Javadoc) * @see org.springframework.beans.factory.DisposableBean#destroy() @@ -118,7 +171,7 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub @Override public void destroy() { - List publications = events.findIncompletePublications(); + var publications = events.findIncompletePublications(); if (publications.isEmpty()) { @@ -130,8 +183,8 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub for (int i = 0; i < publications.size(); i++) { - String prefix = i + 1 == publications.size() ? "└─" : "├─"; - EventPublication it = publications.get(i); + var prefix = i + 1 == publications.size() ? "└─" : "├─"; + var it = publications.get(i); LOGGER.info("{} {} - {}", prefix, it.getEvent().getClass().getName(), it.getTargetIdentifier().getValue()); } 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 cd48b5f7..d05179ff 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 @@ -32,19 +32,29 @@ import org.springframework.context.ApplicationListener; public interface EventPublicationRegistry { /** - * Stores {@link EventPublication}s for the given event and {@link ApplicationListener}s. + * Stores {@link TargetEventPublication}s for the given event and {@link ApplicationListener}s. * * @param event must not be {@literal null}. * @param listeners must not be {@literal null}. */ - Collection store(Object event, Stream listeners); + Collection store(Object event, Stream listeners); /** - * Returns all {@link EventPublication}s that have not been completed yet. + * Returns all {@link TargetEventPublication}s that have not been completed yet. * * @return will never be {@literal null}. */ - Collection findIncompletePublications(); + Collection findIncompletePublications(); + + /** + * Returns all {@link TargetEventPublication}s that have not been completed yet and have been published before the + * given duration in relation to "now". + * + * @param duration must not be {@literal null}. + * @return will never be {@literal null}. + * @since 1.1 + */ + Collection findIncompletePublicationsOlderThan(Duration duration); /** * Marks the publication for the given event and {@link PublicationTargetIdentifier} as completed. @@ -55,7 +65,7 @@ public interface EventPublicationRegistry { void markCompleted(Object event, PublicationTargetIdentifier targetIdentifier); /** - * Deletes all completed {@link EventPublication}s that have been completed before the given {@link Duration}. + * Deletes all completed {@link TargetEventPublication}s that have been completed before the given {@link Duration}. * * @param duration must not be {@literal null}. */ 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 6abda3e2..a1eaa71f 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 @@ -18,11 +18,12 @@ package org.springframework.modulith.events.core; import java.time.Instant; import java.util.List; import java.util.Optional; +import java.util.UUID; import org.springframework.util.Assert; /** - * Repository to store {@link EventPublication}s. + * Repository to store {@link TargetEventPublication}s. * * @author Björn Kieling * @author Dmitry Belyaev @@ -31,20 +32,20 @@ import org.springframework.util.Assert; public interface EventPublicationRepository { /** - * Persists the given {@link EventPublication}. + * Persists the given {@link TargetEventPublication}. * * @param publication must not be {@literal null}. * @return will never be {@literal null}. */ - EventPublication create(EventPublication publication); + TargetEventPublication create(TargetEventPublication publication); /** - * Marks the given {@link EventPublication} as completed. + * Marks the given {@link TargetEventPublication} as completed. * * @param publication must not be {@literal null}. * @param completionDate must not be {@literal null}. */ - default void markCompleted(EventPublication publication, Instant completionDate) { + default void markCompleted(TargetEventPublication publication, Instant completionDate) { Assert.notNull(publication, "EventPublication must not be null!"); Assert.notNull(completionDate, "Instant must not be null!"); @@ -65,22 +66,40 @@ public interface EventPublicationRepository { void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate); /** - * Returns all {@link EventPublication} that have not been completed yet. + * Returns all {@link TargetEventPublication}s that have not been completed yet. * * @return will never be {@literal null}. */ - List findIncompletePublications(); + List findIncompletePublications(); /** - * Return the incomplete {@link EventPublication} for the given serialized event and listener identifier. + * Returns all {@link TargetEventPublication}s that have not been completed and were published before the given + * {@link Instant}. + * + * @param instant must not be {@literal null}. + * @return will never be {@literal null}. + * @since 1.1 + */ + List findIncompletePublicationsPublishedBefore(Instant instant); + + /** + * Return the incomplete {@link TargetEventPublication} 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 findIncompletePublicationsByEventAndTargetIdentifier( // + Optional findIncompletePublicationsByEventAndTargetIdentifier( // Object event, PublicationTargetIdentifier targetIdentifier); + /** + * Deletes all publications with the given identifiers. + * + * @param identifiers must not be {@literal null}. + * @since 1.1 + */ + void deletePublications(List identifiers); + /** * Deletes all publications that were already marked as completed. */ diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/PublicationTargetIdentifier.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/PublicationTargetIdentifier.java index 44bafaf3..c7c67e73 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/PublicationTargetIdentifier.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/PublicationTargetIdentifier.java @@ -17,6 +17,7 @@ package org.springframework.modulith.events.core; import java.util.Objects; +import org.springframework.lang.Nullable; import org.springframework.util.Assert; /** @@ -82,7 +83,7 @@ public class PublicationTargetIdentifier { * @see java.lang.Object#equals(java.lang.Object) */ @Override - public boolean equals(Object obj) { + public boolean equals(@Nullable Object obj) { if (this == obj) { return true; diff --git a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublication.java b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java similarity index 51% rename from spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublication.java rename to spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java index 1db13ed9..f8f2bf32 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublication.java +++ b/spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java @@ -17,11 +17,7 @@ package org.springframework.modulith.events.core; import java.time.Clock; import java.time.Instant; -import java.util.Optional; -import java.util.UUID; -import org.springframework.context.ApplicationEvent; -import org.springframework.context.PayloadApplicationEvent; import org.springframework.util.Assert; /** @@ -31,10 +27,10 @@ import org.springframework.util.Assert; * @author Björn Kieling * @author Dmitry Belyaev */ -public interface EventPublication extends Comparable, Completable { +public interface TargetEventPublication extends Completable, org.springframework.modulith.events.EventPublication { /** - * Creates a {@link EventPublication} for the given event an listener identifier using a default {@link Instant}. + * 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}. * @@ -43,58 +39,22 @@ public interface EventPublication extends Comparable, Completa * @return will never be {@literal null}. * @see #of(Object, PublicationTargetIdentifier, Instant) */ - static EventPublication of(Object event, PublicationTargetIdentifier id) { + static TargetEventPublication of(Object event, PublicationTargetIdentifier id) { return new DefaultEventPublication(event, id, Instant.now()); } /** - * Creates a {@link EventPublication} for the given event an listener identifier and publication date. + * Creates a {@link TargetEventPublication} for the given event an listener identifier and publication date. * * @param event must not be {@literal null}. * @param id must not be {@literal null}. * @param publicationDate must not be {@literal null}. * @return will never be {@literal null}. */ - static EventPublication of(Object event, PublicationTargetIdentifier id, Instant publicationDate) { + static TargetEventPublication of(Object event, PublicationTargetIdentifier id, Instant publicationDate) { return new DefaultEventPublication(event, id, publicationDate); } - /** - * Returns a unique identifier for this publication. - * - * @return will never be {@literal null}. - */ - UUID getIdentifier(); - - /** - * Returns the event that is published. - * - * @return - */ - Object getEvent(); - - /** - * Returns the event as Spring {@link ApplicationEvent}, effectively wrapping it into a - * {@link PayloadApplicationEvent} in case it's not one already. - * - * @return - */ - default ApplicationEvent getApplicationEvent() { - - Object event = getEvent(); - - return PayloadApplicationEvent.class.isInstance(event) // - ? PayloadApplicationEvent.class.cast(event) - : new PayloadApplicationEvent<>(this, event); - } - - /** - * Returns the time the event is published at. - * - * @return - */ - Instant getPublicationDate(); - /** * Returns the identifier of the target that the event is supposed to be published to. * @@ -114,29 +74,4 @@ public interface EventPublication extends Comparable, Completa return this.getTargetIdentifier().equals(identifier); } - - /** - * Returns the completion date of the publication. - * - * @return will never be {@literal null}. - */ - Optional getCompletionDate(); - - /** - * Returns whether the publication of the event has completed. - * - * @return will never be {@literal null}. - */ - default boolean isPublicationCompleted() { - return getCompletionDate().isPresent(); - } - - /* - * (non-Javadoc) - * @see java.lang.Comparable#compareTo(java.lang.Object) - */ - @Override - public default int compareTo(EventPublication that) { - return this.getPublicationDate().compareTo(that.getPublicationDate()); - } } 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 beedebd1..552f32cc 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 @@ -15,9 +15,11 @@ */ package org.springframework.modulith.events.support; +import java.time.Duration; import java.util.Collection; import java.util.List; import java.util.function.Consumer; +import java.util.function.Predicate; import java.util.function.Supplier; import java.util.stream.Stream; @@ -33,9 +35,12 @@ import org.springframework.core.ResolvableType; import org.springframework.core.annotation.AnnotationAwareOrderComparator; import org.springframework.core.env.Environment; import org.springframework.lang.NonNull; -import org.springframework.modulith.events.core.EventPublication; +import org.springframework.lang.Nullable; +import org.springframework.modulith.events.EventPublication; +import org.springframework.modulith.events.IncompleteEventPublications; import org.springframework.modulith.events.core.EventPublicationRegistry; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.transaction.event.TransactionPhase; import org.springframework.transaction.event.TransactionalApplicationListener; import org.springframework.transaction.event.TransactionalEventListener; @@ -53,7 +58,7 @@ import org.springframework.util.Assert; * @see CompletionRegisteringAdvisor */ public class PersistentApplicationEventMulticaster extends AbstractApplicationEventMulticaster - implements SmartInitializingSingleton { + implements IncompleteEventPublications, SmartInitializingSingleton { private static final Logger LOGGER = LoggerFactory.getLogger(PersistentApplicationEventMulticaster.class); static final String REPUBLISH_ON_RESTART = "spring.modulith.republish-outstanding-events-on-restart"; @@ -92,7 +97,7 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv */ @Override @SuppressWarnings({ "unchecked", "rawtypes" }) - public void multicastEvent(ApplicationEvent event, ResolvableType eventType) { + public void multicastEvent(ApplicationEvent event, @Nullable ResolvableType eventType) { var type = eventType == null ? ResolvableType.forInstance(event) : eventType; var listeners = getApplicationListeners(event, type); @@ -109,6 +114,24 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv } } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.IncompleteEventPublications#resubmitIncompletePublications(java.util.function.Predicate) + */ + @Override + public void resubmitIncompletePublications(Predicate filter) { + doResubmitUncompletedPublicationsOlderThan(null); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.IncompleteEventPublications#resubmitIncompletePublicationsOlderThan(java.time.Duration) + */ + @Override + public void resubmitIncompletePublicationsOlderThan(Duration duration) { + doResubmitUncompletedPublicationsOlderThan(duration); + } + /* * (non-Javadoc) * @see org.springframework.beans.factory.SmartInitializingSingleton#afterSingletonsInstantiated() @@ -120,16 +143,10 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv return; } - LOGGER.debug("Looking up previously pending event publications…"); - - var publications = registry.get().findIncompletePublications(); - - LOGGER.debug("{} found.", publications.isEmpty() ? "None" : publications.size()); - - publications.forEach(this::invokeTargetListener); + resubmitIncompletePublications(__ -> true); } - private void invokeTargetListener(EventPublication publication) { + private void invokeTargetListener(TargetEventPublication publication) { var listeners = new TransactionalEventListeners( getApplicationListeners()); @@ -145,7 +162,23 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv }); } - private ApplicationListener executeListenerWithCompletion(EventPublication publication, + private void doResubmitUncompletedPublicationsOlderThan(@Nullable Duration duration) { + + 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.forEach(this::invokeTargetListener); + } + + private static ApplicationListener executeListenerWithCompletion(EventPublication publication, TransactionalApplicationListener listener) { listener.processEvent(publication.getApplicationEvent()); @@ -169,6 +202,17 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv : event; } + 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/EventPublicationUnitTests.java b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/TargetEventPublicationUnitTests.java similarity index 83% rename from spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/EventPublicationUnitTests.java rename to spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/TargetEventPublicationUnitTests.java index 7c54aa7e..b7133952 100644 --- a/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/EventPublicationUnitTests.java +++ b/spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/TargetEventPublicationUnitTests.java @@ -24,13 +24,13 @@ import org.junit.jupiter.api.Test; * @author Björn Kieling * @author Dmitry Belyaev */ -class EventPublicationUnitTests { +class TargetEventPublicationUnitTests { @Test void rejectsNullEvent() { assertThatExceptionOfType(IllegalArgumentException.class)// - .isThrownBy(() -> EventPublication.of(null, PublicationTargetIdentifier.of("foo")))// + .isThrownBy(() -> TargetEventPublication.of(null, PublicationTargetIdentifier.of("foo")))// .withMessageContaining("Event"); } @@ -38,14 +38,14 @@ class EventPublicationUnitTests { void rejectsNullTargetIdentifier() { assertThatExceptionOfType(IllegalArgumentException.class)// - .isThrownBy(() -> EventPublication.of(new Object(), null))// + .isThrownBy(() -> TargetEventPublication.of(new Object(), null))// .withMessageContaining("TargetIdentifier"); } @Test void publicationIsIncompleteByDefault() { - EventPublication publication = EventPublication.of(new Object(), + var publication = TargetEventPublication.of(new Object(), PublicationTargetIdentifier.of("foo")); assertThat(publication.isPublicationCompleted()).isFalse(); 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 fb6fb72f..a4fce816 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 @@ -26,7 +26,7 @@ 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.EventPublication; +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; @@ -64,10 +64,10 @@ class PersistentApplicationEventMulticasterIntegrationTests { void doesNotPublishGenericEventsToListeners() throws Exception { publisher.publishEvent(new SomeGenericEvent<>()); - verify(repository, never()).create(any(EventPublication.class)); + verify(repository, never()).create(any(TargetEventPublication.class)); publisher.publishEvent(new SomeOtherEvent()); - verify(repository).create(any(EventPublication.class)); + verify(repository).create(any(TargetEventPublication.class)); } @Component 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 82eea303..73fdca93 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 @@ -20,10 +20,12 @@ import java.sql.SQLException; import java.sql.Timestamp; import java.time.Instant; import java.util.ArrayList; +import java.util.Collections; import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.UUID; +import java.util.stream.IntStream; import org.slf4j.Logger; import org.slf4j.LoggerFactory; @@ -31,15 +33,15 @@ 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.EventPublication; import org.springframework.modulith.events.core.EventPublicationRepository; import org.springframework.modulith.events.core.EventSerializer; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; /** - * JDBC-based repository to store {@link EventPublication}s. + * JDBC-based repository to store {@link TargetEventPublication}s. * * @author Dmitry Belyaev * @author Björn Kieling @@ -61,10 +63,13 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { ORDER BY PUBLICATION_DATE ASC """; - private static final String SQL_STATEMENT_UPDATE = """ - UPDATE EVENT_PUBLICATION - SET COMPLETION_DATE = ? - WHERE ID = ? + private static final String SQL_STATEMENT_FIND_UNCOMPLETED_BEFORE = """ + SELECT ID, COMPLETION_DATE, EVENT_TYPE, LISTENER_ID, PUBLICATION_DATE, SERIALIZED_EVENT + FROM EVENT_PUBLICATION + WHERE + COMPLETION_DATE IS NULL + AND PUBLICATION_DATE < ? + ORDER BY PUBLICATION_DATE ASC """; private static final String SQL_STATEMENT_UPDATE_BY_EVENT_AND_LISTENER_ID = """ @@ -85,6 +90,13 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { ORDER BY PUBLICATION_DATE """; + private static final String SQL_STATEMENT_DELETE = """ + DELETE + FROM EVENT_PUBLICATION + WHERE + ID IN (?) + """; + private static final String SQL_STATEMENT_DELETE_UNCOMPLETED = """ DELETE FROM EVENT_PUBLICATION @@ -99,6 +111,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { COMPLETION_DATE < ? """; + private static final int DELETE_BATCH_SIZE = 100; + private final JdbcOperations operations; private final EventSerializer serializer; private final DatabaseType databaseType; @@ -129,7 +143,7 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { */ @Override @Transactional - public EventPublication create(EventPublication publication) { + public TargetEventPublication create(TargetEventPublication publication) { var serializedEvent = serializeEvent(publication.getEvent()); @@ -158,25 +172,59 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { serializer.serialize(event)); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier) + */ @Override @Transactional(readOnly = true) - public Optional findIncompletePublicationsByEventAndTargetIdentifier( // + public Optional findIncompletePublicationsByEventAndTargetIdentifier( // Object event, PublicationTargetIdentifier targetIdentifier) { - var serializedEvent = serializeEvent(event); - var listenerId = targetIdentifier.getValue(); + var result = operations.query(SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, // + this::resultSetToPublications, // + serializeEvent(event), // + targetIdentifier.getValue()); - return findAllIncompletePublicationsByEventAndListenerId(serializedEvent, listenerId).stream() // - .findFirst(); + return result == null ? Optional.empty() : result.stream().findFirst(); } @Override @Transactional(readOnly = true) @SuppressWarnings("null") - public List findIncompletePublications() { + public List findIncompletePublications() { return operations.query(SQL_STATEMENT_FIND_UNCOMPLETED, this::resultSetToPublications); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsPublishedBefore(java.time.Instant) + */ + @Override + public List findIncompletePublicationsPublishedBefore(Instant instant) { + + var result = operations.query(SQL_STATEMENT_FIND_UNCOMPLETED_BEFORE, + this::resultSetToPublications, Timestamp.from(instant)); + + return result == null ? Collections.emptyList() : result; + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) + */ + @Override + public void deletePublications(List identifiers) { + + var databaseIds = identifiers.stream().map(this::uuidToDatabase).toList(); + + operations.batchUpdate(SQL_STATEMENT_DELETE, batch(databaseIds, DELETE_BATCH_SIZE)); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deleteCompletedPublications() + */ @Override public void deleteCompletedPublications() { operations.execute(SQL_STATEMENT_DELETE_UNCOMPLETED); @@ -194,31 +242,20 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { operations.update(SQL_STATEMENT_DELETE_UNCOMPLETED_BEFORE, Timestamp.from(instant)); } - @SuppressWarnings("null") - 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) { return serializer.serialize(event).toString(); } /** - * Effectively a {@link ResultSetExtractor} to drop {@link EventPublication}s that cannot be deserialized. + * Effectively a {@link ResultSetExtractor} to drop {@link TargetEventPublication}s that cannot be deserialized. * * @param resultSet must not be {@literal null}. * @return will never be {@literal null}. * @throws SQLException */ - private List resultSetToPublications(ResultSet resultSet) throws SQLException { + private List resultSetToPublications(ResultSet resultSet) throws SQLException { - List result = new ArrayList<>(); + List result = new ArrayList<>(); while (resultSet.next()) { @@ -233,14 +270,14 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { } /** - * Effectively a {@link RowMapper} to turn a single row into an {@link EventPublication}. + * Effectively a {@link RowMapper} to turn a single row into an {@link TargetEventPublication}. * * @param rs must not be {@literal null}. * @return can be {@literal null}. * @throws SQLException */ @Nullable - private EventPublication resultSetToPublication(ResultSet rs) throws SQLException { + private TargetEventPublication resultSetToPublication(ResultSet rs) throws SQLException { var id = getUuidFromResultSet(rs); var eventClass = loadClass(id, rs.getString("EVENT_TYPE")); @@ -277,7 +314,17 @@ class JdbcEventPublicationRepository implements EventPublicationRepository { } } - private static class JdbcEventPublication implements EventPublication { + private static List batch(List input, int batchSize) { + + var inputSize = input.size(); + + return IntStream.range(0, (inputSize + batchSize - 1) / batchSize) + .mapToObj(i -> input.subList(i * batchSize, Math.min((i + 1) * batchSize, inputSize))) + .map(List::toArray) + .toList(); + } + + private static class JdbcEventPublication implements TargetEventPublication { private final UUID id; private final Instant publicationDate; 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 13d37738..f63ea4fd 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 @@ -26,6 +26,7 @@ import java.time.LocalDateTime; import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; import java.util.Comparator; +import java.util.List; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Nested; @@ -35,9 +36,9 @@ import org.springframework.boot.test.autoconfigure.jdbc.JdbcTest; import org.springframework.boot.test.mock.mockito.MockBean; import org.springframework.context.annotation.Import; import org.springframework.jdbc.core.JdbcOperations; -import org.springframework.modulith.events.core.EventPublication; import org.springframework.modulith.events.core.EventSerializer; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.modulith.testapp.TestApplication; import org.springframework.test.context.ActiveProfiles; import org.springframework.test.context.ContextConfiguration; @@ -79,7 +80,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent)).thenReturn(serializedEvent); when(serializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); - var publication = repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + var publication = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER)); var eventPublications = repository.findIncompletePublications(); @@ -110,11 +111,11 @@ class JdbcEventPublicationRepositoryIntegrationTests { createPublicationAt(now.withHour(1)); assertThat(repository.findIncompletePublications()) - .isSortedAccordingTo(Comparator.comparing(EventPublication::getPublicationDate)); + .isSortedAccordingTo(Comparator.comparing(TargetEventPublication::getPublicationDate)); } private void createPublicationAt(LocalDateTime publicationDate) { - repository.create(EventPublication.of("", TARGET_IDENTIFIER, publicationDate.toInstant(ZoneOffset.UTC))); + repository.create(TargetEventPublication.of("", TARGET_IDENTIFIER, publicationDate.toInstant(ZoneOffset.UTC))); } @Test // GH-3 @@ -130,14 +131,14 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent2)).thenReturn(serializedEvent2); when(serializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); - repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - var publication = repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER)); + var publication = repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER)); // Complete publication repository.markCompleted(publication, Instant.now()); assertThat(repository.findIncompletePublications()).hasSize(1) - .element(0).extracting(EventPublication::getEvent).isEqualTo(testEvent1); + .element(0).extracting(TargetEventPublication::getEvent).isEqualTo(testEvent1); } @Test // GH-3 @@ -161,7 +162,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent)).thenReturn(serializedEvent); when(serializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); - var publication = EventPublication.of(testEvent, TARGET_IDENTIFIER); + var publication = TargetEventPublication.of(testEvent, TARGET_IDENTIFIER); repository.create(publication); repository.markCompleted(publication, Instant.now()); @@ -180,9 +181,9 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent)).thenReturn(serializedEvent); when(serializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); - var publication = repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + var publication = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER)); Thread.sleep(10); - repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER)); var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); @@ -202,7 +203,7 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); // Store publication - repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER)); operations.update("UPDATE EVENT_PUBLICATION SET EVENT_TYPE='abc'"); @@ -223,9 +224,9 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent2)).thenReturn(serializedEvent2); when(serializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); - var publication = repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); + var publication = repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER)); - repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER)); repository.markCompleted(publication, Instant.now()); repository.deleteCompletedPublications(); @@ -246,8 +247,8 @@ class JdbcEventPublicationRepositoryIntegrationTests { when(serializer.serialize(testEvent2)).thenReturn(serializedEvent2); when(serializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); - repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER)); var now = Instant.now(); @@ -258,6 +259,50 @@ class JdbcEventPublicationRepositoryIntegrationTests { assertThat(operations.query("SELECT * FROM EVENT_PUBLICATION", (rs, __) -> rs.getString("SERIALIZED_EVENT"))) .hasSize(1).element(0).isEqualTo(serializedEvent2); } + + @Test // GH-294 + void deletesPublicationsByIdentifier() { + + var first = createPublication(new TestEvent("first")); + var second = createPublication(new TestEvent("second")); + + repository.deletePublications(List.of(first.getIdentifier())); + + assertThat(repository.findIncompletePublications()) + .hasSize(1) + .element(0) + .matches(it -> it.getIdentifier().equals(second.getIdentifier())) + .matches(it -> it.getEvent().equals(second.getEvent())); + } + + @Test // GH-294 + void findsPublicationsOlderThanReference() throws Exception { + + var first = createPublication(new TestEvent("first")); + + Thread.sleep(100); + + var now = Instant.now(); + var second = createPublication(new TestEvent("second")); + + assertThat(repository.findIncompletePublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier(), second.getIdentifier()); + + assertThat(repository.findIncompletePublicationsPublishedBefore(now)) + .hasSize(1) + .element(0).extracting(TargetEventPublication::getIdentifier).isEqualTo(first.getIdentifier()); + } + + private TargetEventPublication createPublication(Object event) { + + var token = event.toString(); + + doReturn(token).when(serializer).serialize(event); + doReturn(event).when(serializer).deserialize(token, event.getClass()); + + return repository.create(TargetEventPublication.of(event, TARGET_IDENTIFIER)); + } } @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 34458d97..4ecc417a 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 @@ -22,30 +22,32 @@ import java.util.List; import java.util.Objects; import java.util.Optional; import java.util.UUID; +import java.util.stream.IntStream; -import org.springframework.modulith.events.core.EventPublication; import org.springframework.modulith.events.core.EventPublicationRepository; import org.springframework.modulith.events.core.EventSerializer; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; /** - * Repository to store {@link EventPublication}s. + * Repository to store {@link TargetEventPublication}s. * * @author Oliver Drotbohm * @author Dmitry Belyaev * @author Björn Kieling */ +@Transactional 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 - and p.completionDate is null + where + p.serializedEvent = ?1 + and p.listenerId = ?2 + and p.completionDate is null """; private static String INCOMPLETE = """ @@ -57,6 +59,16 @@ class JpaEventPublicationRepository implements EventPublicationRepository { p.publicationDate asc """; + private static String INCOMPLETE_BEFORE = """ + select p + from JpaEventPublication p + where + p.completionDate is null + and p.publicationDate < ?1 + order by + p.publicationDate asc + """; + private static final String MARK_COMPLETED_BY_EVENT_AND_LISTENER_ID = """ update JpaEventPublication p set p.completionDate = ?3 @@ -64,6 +76,13 @@ class JpaEventPublicationRepository implements EventPublicationRepository { and p.listenerId = ?2 """; + private static final String DELETE = """ + delete + from JpaEventPublication p + where + p.id in ?1 + """; + private static final String DELETE_COMPLETED = """ delete from JpaEventPublication p @@ -78,6 +97,8 @@ class JpaEventPublicationRepository implements EventPublicationRepository { p.completionDate < ?1 """; + private static final int DELETE_BATCH_SIZE = 100; + private final EntityManager entityManager; private final EventSerializer serializer; @@ -102,8 +123,7 @@ class JpaEventPublicationRepository implements EventPublicationRepository { * @see org.springframework.modulith.events.EventPublicationRepository#create(org.springframework.modulith.events.EventPublication) */ @Override - @Transactional - public EventPublication create(EventPublication publication) { + public TargetEventPublication create(TargetEventPublication publication) { entityManager.persist(domainToEntity(publication)); @@ -115,7 +135,6 @@ class JpaEventPublicationRepository implements EventPublicationRepository { * @see org.springframework.modulith.events.EventPublicationRepository#markCompleted(java.lang.Object, org.springframework.modulith.events.PublicationTargetIdentifier, java.time.Instant) */ @Override - @Transactional public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) { entityManager.createQuery(MARK_COMPLETED_BY_EVENT_AND_LISTENER_ID) @@ -131,7 +150,7 @@ class JpaEventPublicationRepository implements EventPublicationRepository { */ @Override @Transactional(readOnly = true) - public List findIncompletePublications() { + public List findIncompletePublications() { return entityManager.createQuery(INCOMPLETE, JpaEventPublication.class) .getResultStream() @@ -139,25 +158,51 @@ class JpaEventPublicationRepository implements EventPublicationRepository { .toList(); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsPublishedBefore(java.time.Instant) + */ + @Override + @Transactional(readOnly = true) + public List findIncompletePublicationsPublishedBefore(Instant instant) { + + return entityManager.createQuery(INCOMPLETE_BEFORE, JpaEventPublication.class) + .setParameter(1, instant) + .getResultStream() + .map(this::entityToDomain) + .toList(); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.PublicationTargetIdentifier) */ @Override @Transactional(readOnly = true) - public Optional findIncompletePublicationsByEventAndTargetIdentifier( // + public Optional findIncompletePublicationsByEventAndTargetIdentifier( // Object event, PublicationTargetIdentifier targetIdentifier) { return findEntityBySerializedEventAndListenerIdAndCompletionDateNull(event, targetIdentifier) .map(this::entityToDomain); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) + */ + @Override + public void deletePublications(List identifiers) { + + batch(identifiers, DELETE_BATCH_SIZE).forEach(it -> { + entityManager.createQuery(DELETE).setParameter(1, identifiers).executeUpdate(); + }); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.EventPublicationRepository#deleteCompletedPublications() */ @Override - @Transactional public void deleteCompletedPublications() { entityManager.createQuery(DELETE_COMPLETED).executeUpdate(); } @@ -171,10 +216,9 @@ class JpaEventPublicationRepository implements EventPublicationRepository { Assert.notNull(instant, "Instant must not be null!"); - var query = entityManager.createQuery(DELETE_COMPLETED_BEFORE); - - query.setParameter(1, instant); - query.executeUpdate(); + entityManager.createQuery(DELETE_COMPLETED_BEFORE) + .setParameter(1, instant) + .executeUpdate(); } private Optional findEntityBySerializedEventAndListenerIdAndCompletionDateNull( // @@ -193,17 +237,26 @@ class JpaEventPublicationRepository implements EventPublicationRepository { return serializer.serialize(event).toString(); } - private JpaEventPublication domainToEntity(EventPublication domain) { + private JpaEventPublication domainToEntity(TargetEventPublication domain) { return new JpaEventPublication(domain.getIdentifier(), domain.getPublicationDate(), domain.getTargetIdentifier().getValue(), serializeEvent(domain.getEvent()), domain.getEvent().getClass()); } - private EventPublication entityToDomain(JpaEventPublication entity) { + private TargetEventPublication entityToDomain(JpaEventPublication entity) { return new JpaEventPublicationAdapter(entity, serializer); } - private static class JpaEventPublicationAdapter implements EventPublication { + private static List> batch(List input, int batchSize) { + + var inputSize = input.size(); + + return IntStream.range(0, (inputSize + batchSize - 1) / batchSize) + .mapToObj(i -> input.subList(i * batchSize, Math.min((i + 1) * batchSize, inputSize))) + .toList(); + } + + private static class JpaEventPublicationAdapter implements TargetEventPublication { private final JpaEventPublication publication; private final EventSerializer serializer; 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 e9b37742..fad9d238 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 @@ -27,6 +27,7 @@ import java.time.Instant; import java.time.LocalDateTime; import java.time.ZoneOffset; import java.util.Comparator; +import java.util.List; import java.util.UUID; import javax.sql.DataSource; @@ -40,9 +41,9 @@ import org.springframework.context.annotation.Import; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabase; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseBuilder; import org.springframework.jdbc.datasource.embedded.EmbeddedDatabaseType; -import org.springframework.modulith.events.core.EventPublication; import org.springframework.modulith.events.core.EventSerializer; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.orm.jpa.JpaTransactionManager; import org.springframework.orm.jpa.LocalContainerEntityManagerFactoryBean; import org.springframework.orm.jpa.SharedEntityManagerCreator; @@ -127,7 +128,7 @@ class JpaEventPublicationRepositoryIntegrationTests { when(eventSerializer.serialize(testEvent)).thenReturn(serializedEvent); when(eventSerializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); - var publication = repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + var publication = repository.create(TargetEventPublication.of(testEvent, TARGET_IDENTIFIER)); var eventPublications = repository.findIncompletePublications(); @@ -162,7 +163,7 @@ class JpaEventPublicationRepositoryIntegrationTests { when(eventSerializer.serialize(testEvent)).thenReturn(serializedEvent); when(eventSerializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent); - var publication = EventPublication.of(testEvent, TARGET_IDENTIFIER); + var publication = TargetEventPublication.of(testEvent, TARGET_IDENTIFIER); repository.create(publication); repository.markCompleted(publication, Instant.now()); @@ -185,8 +186,8 @@ class JpaEventPublicationRepositoryIntegrationTests { when(eventSerializer.serialize(testEvent2)).thenReturn(serializedEvent2); when(eventSerializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); - repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER)); repository.markCompleted(testEvent1, TARGET_IDENTIFIER, Instant.now()); repository.deleteCompletedPublications(); @@ -205,7 +206,7 @@ class JpaEventPublicationRepositoryIntegrationTests { savePublicationAt(now.withHour(1)); assertThat(repository.findIncompletePublications()) - .isSortedAccordingTo(Comparator.comparing(EventPublication::getPublicationDate)); + .isSortedAccordingTo(Comparator.comparing(TargetEventPublication::getPublicationDate)); } @Test // GH-251 @@ -221,8 +222,8 @@ class JpaEventPublicationRepositoryIntegrationTests { when(eventSerializer.serialize(testEvent2)).thenReturn(serializedEvent2); when(eventSerializer.deserialize(serializedEvent2, TestEvent.class)).thenReturn(testEvent2); - repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent1, TARGET_IDENTIFIER)); + repository.create(TargetEventPublication.of(testEvent2, TARGET_IDENTIFIER)); var now = Instant.now(); @@ -235,6 +236,50 @@ class JpaEventPublicationRepositoryIntegrationTests { .element(0).extracting(it -> it.serializedEvent).isEqualTo(serializedEvent2); } + @Test // GH-294 + void deletesPublicationsByIdentifier() { + + var first = createPublication(new TestEvent("first")); + var second = createPublication(new TestEvent("second")); + + repository.deletePublications(List.of(first.getIdentifier())); + + assertThat(repository.findIncompletePublications()) + .hasSize(1) + .element(0) + .matches(it -> it.getIdentifier().equals(second.getIdentifier())) + .matches(it -> it.getEvent().equals(second.getEvent())); + } + + @Test // GH-294 + void findsPublicationsOlderThanReference() throws Exception { + + var first = createPublication(new TestEvent("first")); + + Thread.sleep(100); + + var now = Instant.now(); + var second = createPublication(new TestEvent("second")); + + assertThat(repository.findIncompletePublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier(), second.getIdentifier()); + + assertThat(repository.findIncompletePublicationsPublishedBefore(now)) + .hasSize(1) + .element(0).extracting(TargetEventPublication::getIdentifier).isEqualTo(first.getIdentifier()); + } + + 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)); + } + private void savePublicationAt(LocalDateTime date) { em.persist(new JpaEventPublication(UUID.randomUUID(), date.toInstant(ZoneOffset.UTC), "", "", Object.class)); } 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 99361b3e..a6ffd5b4 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 @@ -27,22 +27,33 @@ import java.util.UUID; import org.springframework.data.domain.Sort; 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.data.mongodb.core.query.Update; import org.springframework.data.util.TypeInformation; -import org.springframework.modulith.events.core.EventPublication; import org.springframework.modulith.events.core.EventPublicationRepository; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; +import org.springframework.transaction.annotation.Transactional; import org.springframework.util.Assert; /** - * Repository to store {@link EventPublication}s in a MongoDB. + * Repository to store {@link TargetEventPublication}s in a MongoDB. * * @author Björn Kieling * @author Dmitry Belyaev * @author Oliver Drotbohm */ +@Transactional class MongoDbEventPublicationRepository implements EventPublicationRepository { + private static final String COMPLETION_DATE = "completionDate"; + private static final String EVENT = "event"; + private static final String ID = "id"; + private static final String LISTENER_ID = "listenerId"; + private static final String PUBLICATION_DATE = "publicationDate"; + + private static final Sort DEFAULT_SORT = Sort.by(PUBLICATION_DATE).ascending(); + private final MongoTemplate mongoTemplate; /** @@ -62,7 +73,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { * @see org.springframework.modulith.events.EventPublicationRepository#create(org.springframework.modulith.events.EventPublication) */ @Override - public EventPublication create(EventPublication publication) { + public TargetEventPublication create(TargetEventPublication publication) { mongoTemplate.save(domainToDocument(publication)); @@ -76,44 +87,62 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { @Override public void markCompleted(Object event, PublicationTargetIdentifier identifier, Instant completionDate) { - var criteria = byEventAndListenerId(event, identifier); - var update = Update.update("completionDate", completionDate); + var update = Update.update(COMPLETION_DATE, completionDate); - mongoTemplate.updateFirst(query(criteria), update, MongoDbEventPublication.class); + mongoTemplate.updateFirst(byEventAndListenerId(event, identifier), update, MongoDbEventPublication.class); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublications() + */ @Override - public List findIncompletePublications() { - - var query = query(where("completionDate").isNull()) - .with(Sort.by("publicationDate").ascending()); - - return mongoTemplate.find(query, MongoDbEventPublication.class).stream() // - .map(this::documentToDomain) // - .toList(); + @Transactional(readOnly = true) + public List findIncompletePublications() { + return readMapped(defaultQuery(where(COMPLETION_DATE).isNull())); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsPublishedBefore(java.time.Instant) + */ @Override - public Optional findIncompletePublicationsByEventAndTargetIdentifier( + @Transactional(readOnly = true) + public List findIncompletePublicationsPublishedBefore(Instant instant) { + return readMapped(defaultQuery(where(COMPLETION_DATE).isNull().and(PUBLICATION_DATE).lt(instant))); + } + + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsByEventAndTargetIdentifier(java.lang.Object, org.springframework.modulith.events.core.PublicationTargetIdentifier) + */ + @Override + @Transactional(readOnly = true) + public Optional findIncompletePublicationsByEventAndTargetIdentifier( Object event, PublicationTargetIdentifier targetIdentifier) { - var documents = findDocumentsByEventAndTargetIdentifierAndCompletionDateNull(event, targetIdentifier); - var results = documents - .stream() // - .map(this::documentToDomain) // - .toList(); + var results = readMapped(byEventAndListenerId(event, targetIdentifier)); // if there are several events with exactly the same payload we return the oldest one first return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0)); } + /* + * (non-Javadoc) + * @see org.springframework.modulith.events.core.EventPublicationRepository#deletePublications(java.util.List) + */ + @Override + public void deletePublications(List identifiers) { + mongoTemplate.remove(query(where(ID).in(identifiers)), MongoDbEventPublication.class); + } + /* * (non-Javadoc) * @see org.springframework.modulith.events.EventPublicationRepository#deleteCompletedPublications() */ @Override public void deleteCompletedPublications() { - mongoTemplate.remove(query(where("completionDate").ne(null)), MongoDbEventPublication.class); + mongoTemplate.remove(query(where(COMPLETION_DATE).ne(null)), MongoDbEventPublication.class); } /* @@ -125,28 +154,28 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { Assert.notNull(instant, "Instant must not be null!"); - mongoTemplate.remove(query(where("completionDate").lt(instant)), MongoDbEventPublication.class); + mongoTemplate.remove(query(where(COMPLETION_DATE).lt(instant)), MongoDbEventPublication.class); } - private List findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( // - Object event, PublicationTargetIdentifier targetIdentifier) { + private List readMapped(Query query) { - var criteria = byEventAndListenerId(event, targetIdentifier); - var query = query(criteria).with(Sort.by("publicationDate").ascending()); - - return mongoTemplate.find(query, MongoDbEventPublication.class); + return mongoTemplate.query(MongoDbEventPublication.class) + .matching(query) + .stream() + .map(MongoDbEventPublicationRepository::documentToDomain) + .toList(); } - private Criteria byEventAndListenerId(Object event, PublicationTargetIdentifier identifier) { + private Query byEventAndListenerId(Object event, PublicationTargetIdentifier identifier) { var eventAsMongoType = mongoTemplate.getConverter().convertToMongoType(event, TypeInformation.OBJECT); - return where("event").is(eventAsMongoType) // - .and("listenerId").is(identifier.getValue()) - .and("completionDate").isNull(); + return defaultQuery(where(EVENT).is(eventAsMongoType) // + .and(LISTENER_ID).is(identifier.getValue()) + .and(COMPLETION_DATE).isNull()); } - private MongoDbEventPublication domainToDocument(EventPublication publication) { + private static MongoDbEventPublication domainToDocument(TargetEventPublication publication) { return new MongoDbEventPublication( // publication.getIdentifier(), // @@ -155,11 +184,15 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository { publication.getEvent()); } - private EventPublication documentToDomain(MongoDbEventPublication document) { + private static TargetEventPublication documentToDomain(MongoDbEventPublication document) { return new MongoDbEventPublicationAdapter(document); } - private static class MongoDbEventPublicationAdapter implements EventPublication { + private static Query defaultQuery(Criteria criteria) { + return query(criteria).with(DEFAULT_SORT); + } + + private static class MongoDbEventPublicationAdapter implements TargetEventPublication { private final MongoDbEventPublication publication; 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 ee5517b0..f7c50b6a 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 @@ -24,6 +24,7 @@ import java.time.LocalDateTime; import java.time.ZoneOffset; import java.time.temporal.ChronoUnit; import java.util.Comparator; +import java.util.List; import java.util.UUID; import org.junit.jupiter.api.AfterEach; @@ -33,14 +34,15 @@ 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.modulith.events.core.EventPublication; import org.springframework.modulith.events.core.PublicationTargetIdentifier; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.modulith.testapp.TestApplication; import org.springframework.test.context.ContextConfiguration; /** * @author Björn Kieling * @author Dmitry Belyaev + * @author Oliver Drotbohm */ @DataMongoTest @ContextConfiguration(classes = TestApplication.class) @@ -65,8 +67,7 @@ class MongoDbEventPublicationRepositoryTest { @Test // GH-4 void shouldPersistAndUpdateEventPublication() { - var testEvent = new TestEvent("abc"); - var publication = repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + var publication = createPublication(new TestEvent("abc")); var eventPublications = repository.findIncompletePublications(); @@ -74,7 +75,7 @@ class MongoDbEventPublicationRepositoryTest { assertThat(eventPublications.get(0).getEvent()).isEqualTo(publication.getEvent()); assertThat(eventPublications.get(0).getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier()); - assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)) + assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(new TestEvent("abc"), TARGET_IDENTIFIER)) .isPresent(); // Complete publication @@ -86,16 +87,14 @@ class MongoDbEventPublicationRepositoryTest { @Test // GH-4 void shouldUpdateSingleEventPublication() { - var testEvent1 = new TestEvent("id1"); - var testEvent2 = new TestEvent("id2"); + var first = createPublication(new TestEvent("id1")); + var second = createPublication(new TestEvent("id2")); - repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - var publication = repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); - - repository.markCompleted(publication, Instant.now()); + repository.markCompleted(second, Instant.now()); assertThat(repository.findIncompletePublications()).hasSize(1) - .element(0).extracting(EventPublication::getEvent).isEqualTo(testEvent1); + .element(0) + .extracting(TargetEventPublication::getEvent).isEqualTo(first.getEvent()); } @Test // GH-133 @@ -108,7 +107,34 @@ class MongoDbEventPublicationRepositoryTest { savePublicationAt(now.withHour(1)); assertThat(repository.findIncompletePublications()) - .isSortedAccordingTo(Comparator.comparing(EventPublication::getPublicationDate)); + .isSortedAccordingTo(Comparator.comparing(TargetEventPublication::getPublicationDate)); + } + + @Test // GH-294 + void findsPublicationsOlderThanReference() throws Exception { + + var first = createPublication(new TestEvent("first")); + + Thread.sleep(100); + + var now = Instant.now(); + var second = createPublication(new TestEvent("second")); + + assertThat(repository.findIncompletePublications()) + .extracting(TargetEventPublication::getIdentifier) + .containsExactly(first.getIdentifier(), second.getIdentifier()); + + assertThat(repository.findIncompletePublicationsPublishedBefore(now)) + .hasSize(1) + .element(0).extracting(TargetEventPublication::getIdentifier).isEqualTo(first.getIdentifier()); + } + + private TargetEventPublication createPublication(Object event) { + return createPublication(event, TARGET_IDENTIFIER); + } + + private TargetEventPublication createPublication(Object event, PublicationTargetIdentifier id) { + return repository.create(TargetEventPublication.of(event, id)); } private void savePublicationAt(LocalDateTime date) { @@ -123,18 +149,17 @@ class MongoDbEventPublicationRepositoryTest { @Test // GH-4 void shouldFindEventPublicationByEventAndTargetIdentifier() { - var testEvent1 = new TestEvent("abc"); - var testEvent2 = new TestEvent("def"); + var first = createPublication(new TestEvent("abc")); + createPublication(new TestEvent("def")); - repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); - repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - repository - .create(EventPublication.of(testEvent1, PublicationTargetIdentifier.of(TARGET_IDENTIFIER.getValue() + "!"))); + var firstEvent = first.getEvent(); - var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent1, TARGET_IDENTIFIER); + createPublication(firstEvent, PublicationTargetIdentifier.of("somethingDifferen")); + + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(firstEvent, TARGET_IDENTIFIER); assertThat(actual).hasValueSatisfying(it -> { - assertThat(it.getEvent()).isEqualTo(testEvent1); + assertThat(it.getEvent()).isEqualTo(firstEvent); assertThat(it.getTargetIdentifier()).isEqualTo(TARGET_IDENTIFIER); }); } @@ -151,15 +176,12 @@ class MongoDbEventPublicationRepositoryTest { @Test void shouldNotReturnCompletedEvents() { - TestEvent testEvent = new TestEvent("abc"); + var publication = createPublication(new TestEvent("abc")); - EventPublication publication = EventPublication.of(testEvent, TARGET_IDENTIFIER); - - // Store publication - repository.create(publication); repository.markCompleted(publication, Instant.now()); - var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(publication.getEvent(), + TARGET_IDENTIFIER); assertThat(actual).isEmpty(); } @@ -167,13 +189,13 @@ class MongoDbEventPublicationRepositoryTest { @Test // GH-4 void shouldReturnTheOldestEventTest() throws InterruptedException { - var testEvent = new TestEvent("id"); + var publication = createPublication(new TestEvent("id")); - var publication = repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); Thread.sleep(10); - repository.create(EventPublication.of(testEvent, TARGET_IDENTIFIER)); + repository.create(publication); - var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER); + var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(publication.getEvent(), + TARGET_IDENTIFIER); assertThat(actual).hasValueSatisfying(it -> // assertThat(it.getPublicationDate()) // @@ -187,38 +209,49 @@ class MongoDbEventPublicationRepositoryTest { @Test // GH-20 void shouldDeleteCompletedEvents() { - var testEvent1 = new TestEvent("abc"); - var testEvent2 = new TestEvent("def"); + var publication = createPublication(new TestEvent("abc")); + var second = createPublication(new TestEvent("def")); - var publication = repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - - repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); repository.markCompleted(publication, Instant.now()); repository.deleteCompletedPublications(); assertThat(mongoTemplate.findAll(MongoDbEventPublication.class)) // .hasSize(1) // - .element(0).extracting(it -> it.event).isEqualTo(testEvent2); + .element(0) // + .extracting(it -> it.event) // + .isEqualTo(second.getEvent()); } @Test // GH-251 void shouldDeleteCompletedEventsBefore() { - var testEvent1 = new TestEvent("abc"); - var testEvent2 = new TestEvent("def"); - - var publication1 = repository.create(EventPublication.of(testEvent1, TARGET_IDENTIFIER)); - var publication2 = repository.create(EventPublication.of(testEvent2, TARGET_IDENTIFIER)); + var first = createPublication(new TestEvent("abc")); + var second = createPublication(new TestEvent("def")); var now = Instant.now(); - repository.markCompleted(publication1, now.minusSeconds(30)); - repository.markCompleted(publication2, now); + repository.markCompleted(first, now.minusSeconds(30)); + repository.markCompleted(second, now); repository.deleteCompletedPublicationsBefore(now.minusSeconds(15)); assertThat(mongoTemplate.findAll(MongoDbEventPublication.class)) // .hasSize(1) // - .element(0).extracting(it -> it.event).isEqualTo(testEvent2); + .element(0).extracting(it -> it.event).isEqualTo(second.getEvent()); + } + + @Test // GH-294 + void deletesPublicationsByIdentifier() { + + var first = createPublication(new TestEvent("first")); + var second = createPublication(new TestEvent("second")); + + repository.deletePublications(List.of(first.getIdentifier())); + + assertThat(repository.findIncompletePublications()) + .hasSize(1) + .element(0) + .matches(it -> it.getIdentifier().equals(second.getIdentifier())) + .matches(it -> it.getEvent().equals(second.getEvent())); } } diff --git a/spring-modulith-events/spring-modulith-events-tests/src/test/java/example/events/PersistentDomainEventIntegrationTest.java b/spring-modulith-events/spring-modulith-events-tests/src/test/java/example/events/PersistentDomainEventIntegrationTest.java index 3e20d021..22ec008b 100644 --- a/spring-modulith-events/spring-modulith-events-tests/src/test/java/example/events/PersistentDomainEventIntegrationTest.java +++ b/spring-modulith-events/spring-modulith-events-tests/src/test/java/example/events/PersistentDomainEventIntegrationTest.java @@ -29,11 +29,11 @@ import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; import org.springframework.context.event.EventListener; import org.springframework.core.env.MapPropertySource; +import org.springframework.modulith.events.IncompleteEventPublications; import org.springframework.modulith.events.config.EnablePersistentDomainEvents; -import org.springframework.modulith.events.core.EventPublication; import org.springframework.modulith.events.core.EventPublicationRegistry; import org.springframework.modulith.events.core.PublicationTargetIdentifier; -import org.springframework.modulith.events.support.PersistentApplicationEventMulticaster; +import org.springframework.modulith.events.core.TargetEventPublication; import org.springframework.scheduling.annotation.Async; import org.springframework.scheduling.annotation.EnableAsync; import org.springframework.transaction.annotation.Transactional; @@ -74,7 +74,7 @@ class PersistentDomainEventIntegrationTest { } finally { assertThat(registry.findIncompletePublications()) // - .extracting(EventPublication::getTargetIdentifier) // + .extracting(TargetEventPublication::getTargetIdentifier) // .extracting(PublicationTargetIdentifier::getValue) // .hasSize(2) // .allSatisfy(id -> { @@ -86,9 +86,8 @@ class PersistentDomainEventIntegrationTest { } - // Simulate application restart with pending publications - PersistentApplicationEventMulticaster multicaster = context.getBean(PersistentApplicationEventMulticaster.class); - multicaster.afterSingletonsInstantiated(); + // Resubmit failed publications + context.getBean(IncompleteEventPublications.class).resubmitIncompletePublications(__ -> true); Thread.sleep(200); diff --git a/spring-modulith-starters/spring-modulith-starter-jdbc/pom.xml b/spring-modulith-starters/spring-modulith-starter-jdbc/pom.xml index b2dad9f5..18285ad4 100644 --- a/spring-modulith-starters/spring-modulith-starter-jdbc/pom.xml +++ b/spring-modulith-starters/spring-modulith-starter-jdbc/pom.xml @@ -26,10 +26,16 @@ + + org.springframework.modulith + spring-modulith-events-api + 1.1.0-SNAPSHOT + org.springframework.modulith spring-modulith-events-core 1.1.0-SNAPSHOT + runtime org.springframework.modulith diff --git a/spring-modulith-starters/spring-modulith-starter-jpa/pom.xml b/spring-modulith-starters/spring-modulith-starter-jpa/pom.xml index 12342fc4..ec989d2d 100644 --- a/spring-modulith-starters/spring-modulith-starter-jpa/pom.xml +++ b/spring-modulith-starters/spring-modulith-starter-jpa/pom.xml @@ -26,10 +26,16 @@ + + org.springframework.modulith + spring-modulith-events-api + 1.1.0-SNAPSHOT + org.springframework.modulith spring-modulith-events-core 1.1.0-SNAPSHOT + runtime org.springframework.modulith diff --git a/spring-modulith-starters/spring-modulith-starter-mongodb/pom.xml b/spring-modulith-starters/spring-modulith-starter-mongodb/pom.xml index 8ce19632..4c59ca86 100644 --- a/spring-modulith-starters/spring-modulith-starter-mongodb/pom.xml +++ b/spring-modulith-starters/spring-modulith-starter-mongodb/pom.xml @@ -25,11 +25,16 @@ - + + org.springframework.modulith + spring-modulith-events-api + 1.1.0-SNAPSHOT + org.springframework.modulith spring-modulith-events-core 1.1.0-SNAPSHOT + runtime org.springframework.modulith diff --git a/src/docs/asciidoc/40-events.adoc b/src/docs/asciidoc/40-events.adoc index 3e6bad1f..8930bd68 100644 --- a/src/docs/asciidoc/40-events.adoc +++ b/src/docs/asciidoc/40-events.adoc @@ -124,6 +124,30 @@ For a more flexible arrangement, `EventPublicationRegistry` exposes a method ` .The transactional event listener arrangement after execution image::event-publication-registry-end.png[] +[[events.managing-publications]] +== Managing Event Publications + +Event publications may need to be managed in a variety of ways during the runtime of an application. +Incomplete publications might have to be re-submitted to the corresponding listeners after a given amount of time. +Completed publications on the other hand, will likely have to be purged from the database or moved into an archive store. +As the needs for that kind of housekeeping strongly vary from application to application, Spring Modulith offers API to deal with both kinds of publications. +That API is available through the `spring-modulith-events-api` artifact, that you can add to your application: + +.Using Spring Modulith Events API artifact +[source, xml, subs="+attributes"] +---- + + org.springframework.modulith + spring-modulith-events-api + {projectVersion} + +---- + +This artifact contains two primary abstractions, that are available to application code as Spring Beans: + +* `CompletedEventPublications` -- This interface allows accessing all completed event publications, and provides API to immediately purge all of them from the database or the completed publications older that a given duration (for example, 1 minute). +* `IncompleteEventPublications`-- This interface allows accessing all incomplete event publications to resubmit either the ones matching a given predicate or older than a given `Duration` relative to the original publishing date. + [[events.publication-repositories]] == Event Publication Repositories diff --git a/src/docs/asciidoc/90-appendix.adoc b/src/docs/asciidoc/90-appendix.adoc index a951e279..75c2f07e 100644 --- a/src/docs/asciidoc/90-appendix.adoc +++ b/src/docs/asciidoc/90-appendix.adoc @@ -67,18 +67,24 @@ a|* `spring-modulith-actuator` (runtime) |`spring-modulith-starter-jdbc` |`compile` a|* `spring-modulith-starter-core` +* `spring-modulith-events-api` +* `spring-modulith-events-core` (runtime) * `spring-modulith-events-jdbc` (runtime) * `spring-modulith-events-jackson` (runtime) |`spring-modulith-starter-jpa` |`compile` a|* `spring-modulith-starter-core` +* `spring-modulith-events-api` +* `spring-modulith-events-core` (runtime) * `spring-modulith-events-jpa` (runtime) * `spring-modulith-events-jackson` (runtime) |`spring-modulith-starter-mongodb` |`compile` a|* `spring-modulith-starter-core` +* `spring-modulith-events-api` +* `spring-modulith-events-core` (runtime) * `spring-modulith-events-mongodb` (runtime) * `spring-modulith-events-jackson` (runtime)