GH-1654 - Fix caching of TransactionalEventListeners.

We now properly cache the TransactionalEventListeners instances *before* filtering them for matching listeners against the event instance as a second step.

Author: odrotbohm

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

@@ -25,6 +25,7 @@
 import java.util.function.Consumer;
 import java.util.function.Predicate;
 import java.util.function.Supplier;
+import java.util.stream.Collectors;
 import java.util.stream.Stream;
 
 import org.jspecify.annotations.Nullable;
@@ -76,7 +77,7 @@ public class PersistentApplicationEventMulticaster extends AbstractApplicationEv
 	static final String REPUBLISH_ON_RESTART = "spring.modulith.events.republish-outstanding-events-on-restart";
 	static final String REPUBLISH_ON_RESTART_LEGACY = "spring.modulith.republish-outstanding-events-on-restart";
 
-	private final Map<ResolvableType, TransactionalEventListeners> listenerCache = new ConcurrentReferenceHashMap<>();
+	private final Map<CacheKey, TransactionalEventListeners> cache = new ConcurrentReferenceHashMap<>();
 	private final Supplier<EventPublicationRegistry> registry;
 	private final Supplier<Environment> environment;
 
@@ -114,35 +115,34 @@ public void multicastEvent(ApplicationEvent event) {
 	public void multicastEvent(ApplicationEvent event, @Nullable ResolvableType eventType) {
 
 		var type = eventType == null ? ResolvableType.forInstance(event) : eventType;
-		var listeners = getApplicationListeners(event, type);
+		var candidates = super.getApplicationListeners(event, type);
 
-		if (listeners.isEmpty()) {
+		if (candidates.isEmpty()) {
 			return;
 		}
 
-		listenerCache
-				.computeIfAbsent(type, __ -> new TransactionalEventListeners(listeners, environment))
-				.ifPresent(it -> storePublications(it, getEventToPersist(event)));
+		var eventToPersist = getEventToPersist(event);
 
-		for (ApplicationListener listener : listeners) {
-			listener.onApplicationEvent(event);
+		// Find all listeners that will need to be invoked
+		var matchingListeners = candidates.stream() //
+				.filter(it -> matches(event, eventToPersist, it)) //
+				.toList();
+
+		if (matchingListeners.isEmpty()) {
+			return;
 		}
-	}
 
-	/*
-	 * (non-Javadoc)
-	 * @see org.springframework.context.event.AbstractApplicationEventMulticaster#getApplicationListeners(org.springframework.context.ApplicationEvent, org.springframework.core.ResolvableType)
-	 */
-	@Override
-	protected Collection<ApplicationListener<?>> getApplicationListeners(ApplicationEvent event,
-			ResolvableType eventType) {
+		// From the candidates find the transactional ones and cache them by source and target type
+		cache.computeIfAbsent(new CacheKey(type, getSourceType(event)),
+				__ -> new TransactionalEventListeners(candidates, environment))
 
-		Object eventToPersist = getEventToPersist(event);
+				// Make sure we honor the by-event instance evaluated conditions
+				.filter(matchingListeners::contains)
+				.ifPresent(stream -> storePublications(stream, eventToPersist));
 
-		return super.getApplicationListeners(event, eventType)
-				.stream()
-				.filter(it -> matches(event, eventToPersist, it))
-				.toList();
+		for (ApplicationListener listener : matchingListeners) {
+			listener.onApplicationEvent(event);
+		}
 	}
 
 	/*
@@ -202,8 +202,7 @@ public void afterSingletonsInstantiated() {
 
 	private void invokeTargetListener(TargetEventPublication publication) {
 
-		var listeners = new TransactionalEventListeners(
-				getApplicationListeners(), environment);
+		var listeners = new TransactionalEventListeners(getApplicationListeners(), environment);
 
 		listeners.stream() //
 				.filter(it -> publication.isIdentifiedBy(PublicationTargetIdentifier.of(it.getListenerId()))) //
@@ -251,6 +250,13 @@ private static Object getEventToPersist(ApplicationEvent event) {
 				: event;
 	}
 
+	private static @Nullable Class<?> getSourceType(ApplicationEvent event) {
+
+		var source = event.getSource();
+
+		return source != null ? source.getClass() : null;
+	}
+
 	private static boolean matches(ApplicationEvent event, Object payload, ApplicationListener<?> listener) {
 
 		// Verify general listener matching by eagerly evaluating the condition
@@ -280,6 +286,13 @@ private static boolean invokeShouldHandle(ApplicationListener<?> candidate, Appl
 		return true;
 	}
 
+	private record CacheKey(ResolvableType eventType, @Nullable Class<?> sourceType) {
+
+		private CacheKey {
+			Assert.notNull(eventType, "Event type must not be null");
+		}
+	}
+
 	/**
 	 * First-class collection to work with transactional event listeners, i.e. {@link ApplicationListener} instances that
 	 * implement {@link TransactionalApplicationListener}.
@@ -324,6 +337,10 @@ public TransactionalEventListeners(Collection<ApplicationListener<?>> listeners,
 					.toList();
 		}
 
+		private TransactionalEventListeners(List<TransactionalApplicationListener<ApplicationEvent>> listeners) {
+			this.listeners = listeners;
+		}
+
 		/**
 		 * Invokes the given {@link Consumer} for all transactional event listeners.
 		 *
@@ -350,6 +367,13 @@ public void ifPresent(Consumer<Stream<TransactionalApplicationListener<Applicati
 			}
 		}
 
+		public TransactionalEventListeners filter(
+				Predicate<? super TransactionalApplicationListener<ApplicationEvent>> filter) {
+
+			return listeners.stream().filter(filter)
+					.collect(Collectors.collectingAndThen(Collectors.toUnmodifiableList(), TransactionalEventListeners::new));
+		}
+
 		/**
 		 * Returns all transactional event listeners.
 		 *

spring-modulith-events/spring-modulith-events-core/src/test/java/org/springframework/modulith/events/support/PersistentApplicationEventMulticasterUnitTests.java

@@ -22,25 +22,28 @@
 
 import java.util.List;
 import java.util.Map;
+import java.util.stream.Stream;
 
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
+import org.mockito.ArgumentCaptor;
 import org.springframework.context.PayloadApplicationEvent;
 import org.springframework.context.annotation.AnnotationConfigApplicationContext;
 import org.springframework.context.event.ApplicationEventMulticaster;
 import org.springframework.context.event.EventListenerMethodProcessor;
-import org.springframework.core.ResolvableType;
 import org.springframework.core.env.MapPropertySource;
 import org.springframework.core.env.StandardEnvironment;
 import org.springframework.mock.env.MockEnvironment;
 import org.springframework.modulith.events.ApplicationModuleListener;
 import org.springframework.modulith.events.core.EventPublicationRegistry;
+import org.springframework.modulith.events.core.PublicationTargetIdentifier;
 import org.springframework.modulith.events.support.PersistentApplicationEventMulticaster.TransactionalEventListeners;
 import org.springframework.stereotype.Component;
 import org.springframework.transaction.event.TransactionPhase;
 import org.springframework.transaction.event.TransactionalApplicationListener;
 import org.springframework.transaction.event.TransactionalApplicationListenerMethodAdapter;
 import org.springframework.transaction.event.TransactionalEventListener;
+import org.springframework.transaction.event.TransactionalEventListenerFactory;
 import org.springframework.util.ReflectionUtils;
 
 /**
@@ -92,21 +95,36 @@ void triggersRepublicationIfLegacyConfigExplicitlyEnabled() {
 		verify(registry).processIncompletePublications(any(), any(), any());
 	}
 
-	@Test // GH-277
+	@Test // GH-277, GH-1654
 	void honorsListenerCondition() throws Exception {
 
 		try (var ctx = new AnnotationConfigApplicationContext()) {
 
 			ctx.addBeanFactoryPostProcessor(new EventListenerMethodProcessor());
+			ctx.registerBean(TransactionalEventListenerFactory.class, TransactionalEventListenerFactory::new);
 			ctx.registerBean("applicationEventMulticaster", ApplicationEventMulticaster.class, () -> multicaster);
-			ctx.registerBean("conditionalListener", ConditionalListener.class);
+			ctx.registerBean("unconditionalListener", UnconditionalListener.class, UnconditionalListener::new);
+			ctx.registerBean("conditionalListener", ConditionalListener.class, ConditionalListener::new);
 			ctx.refresh();
 
-			assertListenerSelected(new SampleEvent(true), true);
-			assertListenerSelected(new SampleEvent(false), false);
+			multicast(new SampleEvent(false));
+			multicast(new SampleEvent(true));
+
+			@SuppressWarnings("unchecked")
+			ArgumentCaptor<Stream<PublicationTargetIdentifier>> captor = ArgumentCaptor.forClass(Stream.class);
+			verify(registry, times(2)).store(any(), captor.capture());
+
+			var allValues = captor.getAllValues();
+
+			assertThat(allValues.get(0).count()).isEqualTo(1L);
+			assertThat(allValues.get(1).count()).isEqualTo(2L);
 		}
 	}
 
+	private void multicast(Object event) {
+		multicaster.multicastEvent(new PayloadApplicationEvent<>(this, event));
+	}
+
 	@Test // GH-726
 	void onlyConsidersAfterCommitListeners() {
 
@@ -170,19 +188,26 @@ private static TransactionalApplicationListenerMethodAdapter getAdapter(Class<?>
 		return new TransactionalApplicationListenerMethodAdapter(type.getName(), type, method);
 	}
 
-	private void assertListenerSelected(SampleEvent event, boolean expected) {
+	@Component
+	static class UnconditionalListener {
 
-		var listeners = multicaster.getApplicationListeners(new PayloadApplicationEvent<>(this, event),
-				ResolvableType.forClass(event.getClass()));
+		boolean invoked = false;
 
-		assertThat(listeners).hasSize(expected ? 1 : 0);
+		@TransactionalEventListener
+		void on(SampleEvent event) {
+			this.invoked = true;
+		}
 	}
 
 	@Component
 	static class ConditionalListener {
 
+		boolean invoked = false;
+
 		@TransactionalEventListener(condition = "#event.supported")
-		void on(SampleEvent event) {}
+		void on(SampleEvent event) {
+			this.invoked = true;
+		}
 	}
 
 	@Component