Spring/spring-modulith
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

GH-294 - Introduce user API to deal with completed and incomplete event publications.

Browse Source 
Introduce a new spring-modulith-events-api artifact to contain types that are supposed to be used by application code to deal with event publications. `EventPublication` was moved into that artifact and got everything non infrastructure related extracted from it's previous incarnation. That in turn has been renamed to `TargetEventPublication`.

`CompletedEventPublications` exposes API to allow purging completed publications either by a given predicate or age (in `Duration`). The interface is implemented by `DefaultEventPublicationRegistry` and thus subject for dependency injection into user code. It primarily delegates to the corresponding methods on `EventPublicationRepository` adapting the given `Duration`s to the `Clock` instance already held internally.

`IncompleteEventPublications` allows triggering the re-submission of incomplete publications by the same criteria as `CEP`. The interface is implemented by `PersistentApplicationEventMulticaster` and this subject for dependency injection into user code.

`EventPublicationRepository` now also allows publications to be deleted by identifiers. The existing implementations have been adapted and batch the requests for every 100 identifiers to prevent a list too large to run into limitations of the underlying data store.

Polished transactional metadata declaration in JPA- and MongoDB-based repository implementations.

Tightened nullability expressions here and there.
This commit is contained in:
Oliver Drotbohm
2023-09-04 09:42:23 +02:00
parent 1fe681bffc
commit 971a143436
29 changed files with 872 additions and 273 deletions
Download Patch File Download Diff File Expand all files Collapse all files

1
spring-modulith-events/pom.xml
View File

@@ -14,6 +14,7 @@
<name>Spring Modulith - Events</name>
<modules>
<module>spring-modulith-events-api</module>
<module>spring-modulith-events-core</module>
<module>spring-modulith-events-jpa</module>
<module>spring-modulith-events-jdbc</module>

35
spring-modulith-events/spring-modulith-events-api/pom.xml Normal file
View File

@@ -0,0 +1,35 @@
<project xmlns="http://maven.apache.org/POM/4.0.0"
xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 https://maven.apache.org/xsd/maven-4.0.0.xsd">
<modelVersion>4.0.0</modelVersion>
<parent>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events</artifactId>
<version>1.1.0-SNAPSHOT</version>
</parent>
<name>Spring Modulith - Events - API</name>
<artifactId>spring-modulith-events-api</artifactId>
<properties>
<module.name>org.springframework.modulith.events.api</module.name>
</properties>
<dependencies>
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>
</dependency>
</dependencies>
</project>

51
spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/CompletedEventPublications.java Normal file
View File

@@ -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<? extends EventPublication> 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<EventPublication> 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);
}

92
spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/EventPublication.java Normal file
View File

@@ -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<Instant> 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());
}
}

44
spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/IncompleteEventPublications.java Normal file
View File

@@ -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 <em>all</em> incomplete event publications.
*
* @param filter a {@link Predicate} to select the event publications for which to resubmit events.
*/
void resubmitIncompletePublications(Predicate<EventPublication> 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);
}

5
spring-modulith-events/spring-modulith-events-api/src/main/java/org/springframework/modulith/events/package-info.java Normal file
View File

@@ -0,0 +1,5 @@
/**
* API of the event publication registry abstraction.
*/
@org.springframework.lang.NonNullApi
package org.springframework.modulith.events;

6
spring-modulith-events/spring-modulith-events-core/pom.xml
View File

@@ -18,6 +18,12 @@
<dependencies>
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-api</artifactId>
<version>${project.version}</version>
</dependency>
<dependency>
<groupId>org.springframework</groupId>
<artifactId>spring-context</artifactId>

2
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/Completable.java
View File

@@ -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
*/

5
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublication.java
View File

@@ -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;

69
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/DefaultEventPublicationRegistry.java
View File

@@ -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<EventPublication> store(Object event, Stream<PublicationTargetIdentifier> listeners) {
public Collection<TargetEventPublication> store(Object event, Stream<PublicationTargetIdentifier> 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<EventPublication> findIncompletePublications() {
public Collection<TargetEventPublication> findIncompletePublications() {
return events.findIncompletePublications();
}
/*
* (non-Javadoc)
* @see org.springframework.modulith.events.core.EventPublicationRegistry#findIncompletePublicationsOlderThan(java.time.Duration)
*/
@Override
public Collection<TargetEventPublication> 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<? extends TargetEventPublication> findAll() {
return findIncompletePublications();
}
/*
* (non-Javadoc)
* @see org.springframework.modulith.events.CompletedEventPublications#deletePublications(java.util.function.Predicate)
*/
@Override
public void deletePublications(Predicate<EventPublication> 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<EventPublication> 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());
}

20
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRegistry.java
View File

@@ -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<EventPublication> store(Object event, Stream<PublicationTargetIdentifier> listeners);
Collection<TargetEventPublication> store(Object event, Stream<PublicationTargetIdentifier> 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<EventPublication> findIncompletePublications();
Collection<TargetEventPublication> 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<TargetEventPublication> 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}.
*/

37
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublicationRepository.java
View File

@@ -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<EventPublication> findIncompletePublications();
List<TargetEventPublication> 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<TargetEventPublication> 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<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
Optional<TargetEventPublication> 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<UUID> identifiers);
/**
* Deletes all publications that were already marked as completed.
*/

3
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/PublicationTargetIdentifier.java
View File

@@ -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;

75
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/EventPublication.java → spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/core/TargetEventPublication.java
View File

@@ -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<EventPublication>, 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<EventPublication>, 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<EventPublication>, Completa
return this.getTargetIdentifier().equals(identifier);
}
/**
* Returns the completion date of the publication.
*
* @return will never be {@literal null}.
*/
Optional<Instant> 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());
}
}

68
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticaster.java
View File

@@ -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<EventPublication> 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<ApplicationEvent> 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<ApplicationEvent> executeListenerWithCompletion(EventPublication publication,
TransactionalApplicationListener<ApplicationEvent> 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}.

8
spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/EventPublicationUnitTests.java → spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/core/TargetEventPublicationUnitTests.java
View File

@@ -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();

6
spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterIntegrationTests.java
View File

@@ -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

107
spring-modulith-events/spring-modulith-events-jdbc/src/main/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepository.java
View File

@@ -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<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
public Optional<TargetEventPublication> 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<EventPublication> findIncompletePublications() {
public List<TargetEventPublication> 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<TargetEventPublication> 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<UUID> 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<EventPublication> 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<EventPublication> resultSetToPublications(ResultSet resultSet) throws SQLException {
private List<TargetEventPublication> resultSetToPublications(ResultSet resultSet) throws SQLException {
List<EventPublication> result = new ArrayList<>();
List<TargetEventPublication> 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<Object[]> 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;

75
spring-modulith-events/spring-modulith-events-jdbc/src/test/java/org/springframework/modulith/events/jdbc/JdbcEventPublicationRepositoryIntegrationTests.java
View File

@@ -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

91
spring-modulith-events/spring-modulith-events-jpa/src/main/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepository.java
View File

@@ -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<EventPublication> findIncompletePublications() {
public List<TargetEventPublication> 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<TargetEventPublication> 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<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
public Optional<TargetEventPublication> 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<UUID> 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<JpaEventPublication> 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 <T> List<List<T>> batch(List<T> 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;

61
spring-modulith-events/spring-modulith-events-jpa/src/test/java/org/springframework/modulith/events/jpa/JpaEventPublicationRepositoryIntegrationTests.java
View File

@@ -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));
}

103
spring-modulith-events/spring-modulith-events-mongodb/src/main/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepository.java
View File

@@ -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<EventPublication> 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<TargetEventPublication> findIncompletePublications() {
return readMapped(defaultQuery(where(COMPLETION_DATE).isNull()));
}
/*
* (non-Javadoc)
* @see org.springframework.modulith.events.core.EventPublicationRepository#findIncompletePublicationsPublishedBefore(java.time.Instant)
*/
@Override
public Optional<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier(
@Transactional(readOnly = true)
public List<TargetEventPublication> 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<TargetEventPublication> 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<UUID> 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<MongoDbEventPublication> findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( //
Object event, PublicationTargetIdentifier targetIdentifier) {
private List<TargetEventPublication> 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;

121
spring-modulith-events/spring-modulith-events-mongodb/src/test/java/org/springframework/modulith/events/mongodb/MongoDbEventPublicationRepositoryTest.java
View File

@@ -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()));
}
}

11
spring-modulith-events/spring-modulith-events-tests/src/test/java/example/events/PersistentDomainEventIntegrationTest.java
View File

@@ -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);

6
spring-modulith-starters/spring-modulith-starter-jdbc/pom.xml
View File

@@ -26,10 +26,16 @@
<!-- Events -->
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-api</artifactId>
<version>1.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-core</artifactId>
<version>1.1.0-SNAPSHOT</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.modulith</groupId>

6
spring-modulith-starters/spring-modulith-starter-jpa/pom.xml
View File

@@ -26,10 +26,16 @@
<!-- Events -->
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-api</artifactId>
<version>1.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-core</artifactId>
<version>1.1.0-SNAPSHOT</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.modulith</groupId>

7
spring-modulith-starters/spring-modulith-starter-mongodb/pom.xml
View File

@@ -25,11 +25,16 @@
</dependency>
<!-- Events -->
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-api</artifactId>
<version>1.1.0-SNAPSHOT</version>
</dependency>
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-core</artifactId>
<version>1.1.0-SNAPSHOT</version>
<scope>runtime</scope>
</dependency>
<dependency>
<groupId>org.springframework.modulith</groupId>

24
src/docs/asciidoc/40-events.adoc
View File

@@ -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"]
----
<dependency>
<groupId>org.springframework.modulith</groupId>
<artifactId>spring-modulith-events-api</artifactId>
<version>{projectVersion}</version>
</dependency>
----
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

6
src/docs/asciidoc/90-appendix.adoc
View File

@@ -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)