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

GH-25 - Fix #findByEventAndTargetIdentifier in all repositories to return only incomplete EventPublications. Rename the method accordingly.

Browse Source 
Signed-off-by: Björn Kieling <bkieling@vmware.com>
This commit is contained in:
Dmitry Belyaev
2022-07-27 10:53:16 +02:00
committed by Oliver Drotbohm
parent 3916082d6f
commit 97a5051930
9 changed files with 122 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files

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

@@ -61,7 +61,7 @@ public class DefaultEventPublicationRegistry implements DisposableBean, EventPub
Assert.notNull(event, "Domain event must not be null!");
Assert.notNull(targetIdentifier, "Listener identifier must not be null!");
events.findByEventAndTargetIdentifier(event, targetIdentifier) //
events.findIncompletePublicationsByEventAndTargetIdentifier(event, targetIdentifier) //
.map(DefaultEventPublicationRegistry::logCompleted) //
.map(e -> CompletableEventPublication.of(e.getEvent(), e.getTargetIdentifier()))
.ifPresent(it -> events.update(it.markCompleted()));

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

@@ -51,11 +51,12 @@ public interface EventPublicationRepository {
List<EventPublication> findIncompletePublications();
/**
* Return the {@link EventPublication} for the given serialized event and listener identifier.
* Return the incomplete {@link EventPublication} for the given serialized event and listener identifier.
*
* @param event must not be {@literal null}.
* @param targetIdentifier must not be {@literal null}.
* @return will never be {@literal null}.
*/
Optional<EventPublication> findByEventAndTargetIdentifier(Object event, PublicationTargetIdentifier targetIdentifier);
Optional<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
Object event, PublicationTargetIdentifier targetIdentifier);
}

4
spring-modulith-events/spring-modulith-events-core/src/main/java/org/springframework/modulith/events/support/MapEventPublicationRepository.java
View File

@@ -69,8 +69,8 @@ public class MapEventPublicationRepository implements EventPublicationRepository
}
@Override
public Optional<EventPublication> findByEventAndTargetIdentifier(Object event,
PublicationTargetIdentifier targetIdentifier) {
public Optional<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier(
Object event, PublicationTargetIdentifier targetIdentifier) {
return Optional.ofNullable(events.get(new Key(event, targetIdentifier)));
}

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

@@ -67,7 +67,10 @@ class JdbcEventPublicationRepository implements EventPublicationRepository {
""";
private static final String SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID = """
SELECT * FROM EVENT_PUBLICATION
WHERE SERIALIZED_EVENT = ? AND LISTENER_ID = ?
WHERE
SERIALIZED_EVENT = ?
AND LISTENER_ID = ?
AND COMPLETION_DATE IS NULL
ORDER BY PUBLICATION_DATE
""";
@@ -109,8 +112,8 @@ class JdbcEventPublicationRepository implements EventPublicationRepository {
@Override
@Transactional(readOnly = true)
public Optional<EventPublication> findByEventAndTargetIdentifier(Object event,
PublicationTargetIdentifier targetIdentifier) {
public Optional<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
Object event, PublicationTargetIdentifier targetIdentifier) {
var results = operations.query(SQL_STATEMENT_FIND_BY_EVENT_AND_LISTENER_ID, this::resultSetToPublications,
serializeEvent(event), targetIdentifier.getValue());

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

@@ -86,7 +86,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
assertThat(it.getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier());
});
assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER))
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER))
.isPresent();
// Complete publication
@@ -130,8 +130,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
@Nested
class FindByEventAndTargetIdentifier {
@Test
// GH-3
@Test // GH-3
void shouldTolerateEmptyResult() {
var testEvent = new TestEvent("id");
@@ -139,11 +138,29 @@ class JdbcEventPublicationRepositoryIntegrationTests {
when(serializer.serialize(testEvent)).thenReturn(serializedEvent);
assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
}
@Test
// GH-3
@Test // GH-3
void shouldNotReturnCompletedEvents() {
var testEvent = new TestEvent("id1");
var serializedEvent = "{\"eventId\":\"id1\"}";
when(serializer.serialize(testEvent)).thenReturn(serializedEvent);
when(serializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent);
var publication = CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER);
repository.create(publication);
repository.update(publication.markCompleted());
var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
assertThat(actual).isEmpty();
}
@Test // GH-3 //
void shouldReturnTheOldestEvent() throws Exception {
var testEvent = new TestEvent("id");
@@ -159,7 +176,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
repository.create(publicationNew);
repository.create(publicationOld);
var actual = repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
assertThat(actual).hasValueSatisfying(it -> {
assertThat(it.getPublicationDate()) //
@@ -182,7 +199,7 @@ class JdbcEventPublicationRepositoryIntegrationTests {
operations.update("UPDATE EVENT_PUBLICATION SET EVENT_TYPE='abc'");
assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
}
}
}

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

@@ -40,8 +40,21 @@ import org.springframework.transaction.annotation.Transactional;
@RequiredArgsConstructor
public class JpaEventPublicationRepository implements EventPublicationRepository {
private static String BY_EVENT_AND_LISTENER_ID = "select p from JpaEventPublication p where p.serializedEvent = ?1 and p.listenerId = ?2";
private static String INCOMPLETE = "select p from JpaEventPublication p where p.completionDate is null";
private static String BY_EVENT_AND_LISTENER_ID = """
select p
from JpaEventPublication p
where
p.serializedEvent = ?1
and p.listenerId = ?2
and p.completionDate is null
""";
private static String INCOMPLETE = """
select p
from JpaEventPublication p
where
p.completionDate is null
""";
private final EntityManager entityManager;
private final EventSerializer serializer;
@@ -62,8 +75,8 @@ public class JpaEventPublicationRepository implements EventPublicationRepository
var id = publication.getTargetIdentifier().getValue();
var event = publication.getEvent();
findEntityBySerializedEventAndListenerId(event, id) //
.setCompletionDate(publication.getCompletionDate().orElse(null));
findEntityBySerializedEventAndListenerIdAndCompletionDateNull(event, id) //
.ifPresent(entity -> entity.setCompletionDate(publication.getCompletionDate().orElse(null)));
return publication;
}
@@ -80,14 +93,15 @@ public class JpaEventPublicationRepository implements EventPublicationRepository
@Override
@Transactional(readOnly = true)
public Optional<EventPublication> findByEventAndTargetIdentifier(Object event,
PublicationTargetIdentifier targetIdentifier) {
public Optional<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier( //
Object event, PublicationTargetIdentifier targetIdentifier) {
return Optional.ofNullable(findEntityBySerializedEventAndListenerId(event, targetIdentifier.getValue()))
return findEntityBySerializedEventAndListenerIdAndCompletionDateNull(event, targetIdentifier.getValue())
.map(this::entityToDomain);
}
private JpaEventPublication findEntityBySerializedEventAndListenerId(Object event, String listenerId) {
private Optional<JpaEventPublication> findEntityBySerializedEventAndListenerIdAndCompletionDateNull( //
Object event, String listenerId) {
var serializedEvent = serializeEvent(event);
@@ -95,7 +109,7 @@ public class JpaEventPublicationRepository implements EventPublicationRepository
.setParameter(1, serializedEvent)
.setParameter(2, listenerId);
return query.getSingleResult();
return query.getResultStream().findFirst();
}
private String serializeEvent(Object event) {

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

@@ -127,7 +127,7 @@ class JpaEventPublicationRepositoryIntegrationTests {
assertThat(eventPublications).hasSize(1);
assertThat(eventPublications.get(0).getEvent()).isEqualTo(publication.getEvent());
assertThat(eventPublications.get(0).getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier());
assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent();
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent();
// Complete publication
repository.update(publication.markCompleted());
@@ -135,6 +135,37 @@ class JpaEventPublicationRepositoryIntegrationTests {
assertThat(repository.findIncompletePublications()).isEmpty();
}
@Test
void shouldTolerateEmptyResult() {
var testEvent = new TestEvent("id");
var serializedEvent = "{\"eventId\":\"id\"}";
when(eventSerializer.serialize(testEvent)).thenReturn(serializedEvent);
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
}
@Test
void shouldNotReturnCompletedEvents() {
TestEvent testEvent = new TestEvent("abc");
String serializedEvent = "{\"eventId\":\"abc\"}";
when(eventSerializer.serialize(testEvent)).thenReturn(serializedEvent);
when(eventSerializer.deserialize(serializedEvent, TestEvent.class)).thenReturn(testEvent);
CompletableEventPublication publication = CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER);
// Store publication
repository.create(publication);
repository.update(publication.markCompleted());
var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
assertThat(actual).isEmpty();
}
@Value
private static final class TestEvent {
String eventId;

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

@@ -68,7 +68,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository {
@Override
public EventPublication update(CompletableEventPublication publication) {
return findDocumentsByEventAndTargetIdentifier(publication.getEvent(), publication.getTargetIdentifier())
return findDocumentsByEventAndTargetIdentifierAndCompletionDateNull(publication.getEvent(), publication.getTargetIdentifier())
.stream()
.findFirst()
.map(document -> document.setCompletionDate(publication.getCompletionDate().orElse(null)))
@@ -88,10 +88,11 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository {
}
@Override
public Optional<EventPublication> findByEventAndTargetIdentifier(
public Optional<EventPublication> findIncompletePublicationsByEventAndTargetIdentifier(
Object event, PublicationTargetIdentifier targetIdentifier) {
var documents = findDocumentsByEventAndTargetIdentifier(event, targetIdentifier);
var documents = findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( //
event, targetIdentifier);
var results = documents
.stream() //
.map(this::documentToDomain) //
@@ -101,7 +102,7 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository {
return results.isEmpty() ? Optional.empty() : Optional.of(results.get(0));
}
private List<MongoDbEventPublication> findDocumentsByEventAndTargetIdentifier(
private List<MongoDbEventPublication> findDocumentsByEventAndTargetIdentifierAndCompletionDateNull( //
Object event, PublicationTargetIdentifier targetIdentifier) {
// we need to enforce writing of the type information
@@ -109,7 +110,8 @@ class MongoDbEventPublicationRepository implements EventPublicationRepository {
var query = Query //
.query(Criteria //
.where("event").is(eventAsMongoType) //
.and("listenerId").is(targetIdentifier.getValue())) //
.and("listenerId").is(targetIdentifier.getValue()) //
.and("completionDate").isNull()) //
.with(Sort.by("publicationDate").ascending());
return mongoTemplate.find(query, MongoDbEventPublication.class);

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

@@ -74,7 +74,7 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb {
assertThat(eventPublications.get(0).getEvent()).isEqualTo(publication.getEvent());
assertThat(eventPublications.get(0).getTargetIdentifier()).isEqualTo(publication.getTargetIdentifier());
assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent();
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isPresent();
// Complete publication
repository.update(publication.markCompleted());
@@ -113,7 +113,7 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb {
repository.create(CompletableEventPublication.of(
testEvent1, PublicationTargetIdentifier.of(TARGET_IDENTIFIER.getValue() + "!")));
var actual = repository.findByEventAndTargetIdentifier(testEvent1, TARGET_IDENTIFIER);
var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent1, TARGET_IDENTIFIER);
assertThat(actual).hasValueSatisfying(it -> {
assertThat(it.getEvent()).isEqualTo(testEvent1);
@@ -126,7 +126,23 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb {
var testEvent = new TestEvent("id");
assertThat(repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
assertThat(repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER)).isEmpty();
}
@Test
void shouldNotReturnCompletedEvents() {
TestEvent testEvent = new TestEvent("abc");
CompletableEventPublication publication = CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER);
// Store publication
repository.create(publication);
repository.update(publication.markCompleted());
var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
assertThat(actual).isEmpty();
}
@Test // GH-4
@@ -139,12 +155,11 @@ class MongoDbEventPublicationRepositoryTest extends WithEmbeddedMongoDb {
Thread.sleep(10);
repository.create(CompletableEventPublication.of(testEvent, TARGET_IDENTIFIER));
var actual = repository.findByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
var actual = repository.findIncompletePublicationsByEventAndTargetIdentifier(testEvent, TARGET_IDENTIFIER);
assertThat(actual).hasValueSatisfying(it -> {
assertThat(it.getPublicationDate()) //
.isCloseTo(publicationOld.getPublicationDate(), within(1, ChronoUnit.MILLIS));
});
assertThat(actual).hasValueSatisfying(it -> //
assertThat(it.getPublicationDate()) //
.isCloseTo(publicationOld.getPublicationDate(), within(1, ChronoUnit.MILLIS)));
}
}