Skip to content

Commit 450cc21

Browse files
jhoellerjhoeller
committed
Support for transactional listeners with reactive transactions
TransactionalApplicationListener and TransactionalEventListener automatically detect a reactive TransactionContext as the event source and register the synchronization accordingly. TransactionalEventPublisher is a convenient delegate for publishing corresponding events with the current TransactionContext as event source. This can also serve as a guideline for similar reactive event purposes. Closes gh-27515 Closes gh-21025 Closes gh-30244
1 parent a9d100e commit 450cc21

File tree

10 files changed

+787
-64
lines changed

10 files changed

+787
-64
lines changed

‎framework-docs/modules/ROOT/pages/data-access/transaction/event.adoc

Lines changed: 8 additions & 4 deletions
Original file line numberDiff line numberDiff line change
@@ -57,10 +57,14 @@ attribute of the annotation to `true`.
5757

5858
[NOTE]
5959
====
60-
`@TransactionalEventListener` only works with thread-bound transactions managed by
61-
`PlatformTransactionManager`. A reactive transaction managed by `ReactiveTransactionManager`
62-
uses the Reactor context instead of thread-local attributes, so from the perspective of
63-
an event listener, there is no compatible active transaction that it can participate in.
60+
As of 6.1, `@TransactionalEventListener` can work with thread-bound transactions managed by
61+
`PlatformTransactionManager` as well as reactive transactions managed by `ReactiveTransactionManager`.
62+
For the former, listeners are guaranteed to see the current thread-bound transaction.
63+
Since the latter uses the Reactor context instead of thread-local variables, the transaction
64+
context needs to be included in the published event instance as the event source.
65+
See the
66+
{api-spring-framework}/transaction/reactive/TransactionalEventPublisher.html[`TransactionalEventPublisher`]
67+
javadoc for details.
6468
====
6569

6670

‎spring-context/src/main/java/org/springframework/context/ApplicationEventPublisher.java

Lines changed: 20 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2019 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -29,6 +29,7 @@
2929
* @see org.springframework.context.ApplicationEvent
3030
* @see org.springframework.context.event.ApplicationEventMulticaster
3131
* @see org.springframework.context.event.EventPublicationInterceptor
32+
* @see org.springframework.transaction.event.TransactionalApplicationListener
3233
*/
3334
@FunctionalInterface
3435
public interface ApplicationEventPublisher {
@@ -42,8 +43,21 @@ public interface ApplicationEventPublisher {
4243
* or even immediate execution at all. Event listeners are encouraged
4344
* to be as efficient as possible, individually using asynchronous
4445
* execution for longer-running and potentially blocking operations.
46+
* <p>For usage in a reactive call stack, include event publication
47+
* as a simple hand-off:
48+
* {@code Mono.fromRunnable(() -> eventPublisher.publishEvent(...))}.
49+
* As with any asynchronous execution, thread-local data is not going
50+
* to be available for reactive listener methods. All state which is
51+
* necessary to process the event needs to be included in the event
52+
* instance itself.
53+
* <p>For the convenient inclusion of the current transaction context
54+
* in a reactive hand-off, consider using
55+
* {@link org.springframework.transaction.reactive.TransactionalEventPublisher#publishEvent(Function)}.
56+
* For thread-bound transactions, this is not necessary since the
57+
* state will be implicitly available through thread-local storage.
4558
* @param event the event to publish
4659
* @see #publishEvent(Object)
60+
* @see ApplicationListener#supportsAsyncExecution()
4761
* @see org.springframework.context.event.ContextRefreshedEvent
4862
* @see org.springframework.context.event.ContextClosedEvent
4963
*/
@@ -61,6 +75,11 @@ default void publishEvent(ApplicationEvent event) {
6175
* or even immediate execution at all. Event listeners are encouraged
6276
* to be as efficient as possible, individually using asynchronous
6377
* execution for longer-running and potentially blocking operations.
78+
* <p>For the convenient inclusion of the current transaction context
79+
* in a reactive hand-off, consider using
80+
* {@link org.springframework.transaction.reactive.TransactionalEventPublisher#publishEvent(Object)}.
81+
* For thread-bound transactions, this is not necessary since the
82+
* state will be implicitly available through thread-local storage.
6483
* @param event the event to publish
6584
* @since 4.2
6685
* @see #publishEvent(ApplicationEvent)

‎spring-tx/src/main/java/org/springframework/transaction/event/TransactionalApplicationListener.java

Lines changed: 18 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2021 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -32,12 +32,13 @@
3232
* allows you to prioritize that listener amongst other listeners running before or after
3333
* transaction completion.
3434
*
35-
* <p><b>NOTE: Transactional event listeners only work with thread-bound transactions
36-
* managed by a {@link org.springframework.transaction.PlatformTransactionManager
37-
* PlatformTransactionManager}.</b> A reactive transaction managed by a
38-
* {@link org.springframework.transaction.ReactiveTransactionManager ReactiveTransactionManager}
39-
* uses the Reactor context instead of thread-local variables, so from the perspective of
40-
* an event listener, there is no compatible active transaction that it can participate in.
35+
* <p>As of 6.1, transactional event listeners can work with thread-bound transactions managed
36+
* by a {@link org.springframework.transaction.PlatformTransactionManager} as well as reactive
37+
* transactions managed by a {@link org.springframework.transaction.ReactiveTransactionManager}.
38+
* For the former, listeners are guaranteed to see the current thread-bound transaction.
39+
* Since the latter uses the Reactor context instead of thread-local variables, the transaction
40+
* context needs to be included in the published event instance as the event source:
41+
* see {@link org.springframework.transaction.reactive.TransactionalEventPublisher}.
4142
*
4243
* @author Juergen Hoeller
4344
* @author Oliver Drotbohm
@@ -60,6 +61,16 @@ default int getOrder() {
6061
return Ordered.LOWEST_PRECEDENCE;
6162
}
6263

64+
/**
65+
* Transaction-synchronized listeners do not support asynchronous execution,
66+
* only their target listener ({@link #processEvent}) potentially does.
67+
* @since 6.1
68+
*/
69+
@Override
70+
default boolean supportsAsyncExecution() {
71+
return false;
72+
}
73+
6374
/**
6475
* Return an identifier for the listener to be able to refer to it individually.
6576
* <p>It might be necessary for specific completion callback implementations

‎spring-tx/src/main/java/org/springframework/transaction/event/TransactionalApplicationListenerAdapter.java

Lines changed: 2 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2020 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -22,7 +22,6 @@
2222
import org.springframework.context.ApplicationEvent;
2323
import org.springframework.context.ApplicationListener;
2424
import org.springframework.core.Ordered;
25-
import org.springframework.transaction.support.TransactionSynchronizationManager;
2625
import org.springframework.util.Assert;
2726

2827
/**
@@ -128,11 +127,7 @@ public void processEvent(E event) {
128127

129128
@Override
130129
public void onApplicationEvent(E event) {
131-
if (TransactionSynchronizationManager.isSynchronizationActive() &&
132-
TransactionSynchronizationManager.isActualTransactionActive()) {
133-
TransactionSynchronizationManager.registerSynchronization(
134-
new TransactionalApplicationListenerSynchronization<>(event, this, this.callbacks));
135-
}
130+
TransactionalApplicationListenerSynchronization.register(event, this, this.callbacks);
136131
}
137132

138133
}

‎spring-tx/src/main/java/org/springframework/transaction/event/TransactionalApplicationListenerMethodAdapter.java

Lines changed: 4 additions & 5 deletions
Original file line numberDiff line numberDiff line change
@@ -25,7 +25,6 @@
2525
import org.springframework.context.event.EventListener;
2626
import org.springframework.context.event.GenericApplicationListener;
2727
import org.springframework.core.annotation.AnnotatedElementUtils;
28-
import org.springframework.transaction.support.TransactionSynchronizationManager;
2928
import org.springframework.util.Assert;
3029

3130
/**
@@ -87,10 +86,10 @@ public void addCallback(SynchronizationCallback callback) {
8786

8887
@Override
8988
public void onApplicationEvent(ApplicationEvent event) {
90-
if (TransactionSynchronizationManager.isSynchronizationActive() &&
91-
TransactionSynchronizationManager.isActualTransactionActive()) {
92-
TransactionSynchronizationManager.registerSynchronization(
93-
new TransactionalApplicationListenerSynchronization<>(event, this, this.callbacks));
89+
if (TransactionalApplicationListenerSynchronization.register(event, this, this.callbacks)) {
90+
if (logger.isDebugEnabled()) {
91+
logger.debug("Registered transaction synchronization for " + event);
92+
}
9493
}
9594
else if (this.annotation.fallbackExecution()) {
9695
if (this.annotation.phase() == TransactionPhase.AFTER_ROLLBACK && logger.isWarnEnabled()) {

‎spring-tx/src/main/java/org/springframework/transaction/event/TransactionalApplicationListenerSynchronization.java

Lines changed: 100 additions & 25 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2020 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -18,19 +18,21 @@
1818

1919
import java.util.List;
2020

21+
import reactor.core.publisher.Mono;
22+
2123
import org.springframework.context.ApplicationEvent;
22-
import org.springframework.transaction.support.TransactionSynchronization;
24+
import org.springframework.core.Ordered;
25+
import org.springframework.transaction.reactive.TransactionContext;
2326

2427
/**
25-
* {@link TransactionSynchronization} implementation for event processing with a
28+
* {@code TransactionSynchronization} implementations for event processing with a
2629
* {@link TransactionalApplicationListener}.
2730
*
2831
* @author Juergen Hoeller
2932
* @since 5.3
3033
* @param <E> the specific {@code ApplicationEvent} subclass to listen to
3134
*/
32-
class TransactionalApplicationListenerSynchronization<E extends ApplicationEvent>
33-
implements TransactionSynchronization {
35+
abstract class TransactionalApplicationListenerSynchronization<E extends ApplicationEvent> implements Ordered {
3436

3537
private final E event;
3638

@@ -53,28 +55,11 @@ public int getOrder() {
5355
return this.listener.getOrder();
5456
}
5557

56-
@Override
57-
public void beforeCommit(boolean readOnly) {
58-
if (this.listener.getTransactionPhase() == TransactionPhase.BEFORE_COMMIT) {
59-
processEventWithCallbacks();
60-
}
61-
}
62-
63-
@Override
64-
public void afterCompletion(int status) {
65-
TransactionPhase phase = this.listener.getTransactionPhase();
66-
if (phase == TransactionPhase.AFTER_COMMIT && status == STATUS_COMMITTED) {
67-
processEventWithCallbacks();
68-
}
69-
else if (phase == TransactionPhase.AFTER_ROLLBACK && status == STATUS_ROLLED_BACK) {
70-
processEventWithCallbacks();
71-
}
72-
else if (phase == TransactionPhase.AFTER_COMPLETION) {
73-
processEventWithCallbacks();
74-
}
58+
public TransactionPhase getTransactionPhase() {
59+
return this.listener.getTransactionPhase();
7560
}
7661

77-
private void processEventWithCallbacks() {
62+
public void processEventWithCallbacks() {
7863
this.callbacks.forEach(callback -> callback.preProcessEvent(this.event));
7964
try {
8065
this.listener.processEvent(this.event);
@@ -86,4 +71,94 @@ private void processEventWithCallbacks() {
8671
this.callbacks.forEach(callback -> callback.postProcessEvent(this.event, null));
8772
}
8873

74+
75+
public static <E extends ApplicationEvent> boolean register(
76+
E event, TransactionalApplicationListener<E> listener,
77+
List<TransactionalApplicationListener.SynchronizationCallback> callbacks) {
78+
79+
if (org.springframework.transaction.support.TransactionSynchronizationManager.isSynchronizationActive() &&
80+
org.springframework.transaction.support.TransactionSynchronizationManager.isActualTransactionActive()) {
81+
org.springframework.transaction.support.TransactionSynchronizationManager.registerSynchronization(
82+
new PlatformSynchronization<>(event, listener, callbacks));
83+
return true;
84+
}
85+
else if (event.getSource() instanceof TransactionContext txContext) {
86+
org.springframework.transaction.reactive.TransactionSynchronizationManager rtsm =
87+
new org.springframework.transaction.reactive.TransactionSynchronizationManager(txContext);
88+
if (rtsm.isSynchronizationActive() && rtsm.isActualTransactionActive()) {
89+
rtsm.registerSynchronization(new ReactiveSynchronization<>(event, listener, callbacks));
90+
return true;
91+
}
92+
}
93+
return false;
94+
}
95+
96+
97+
private static class PlatformSynchronization<AE extends ApplicationEvent>
98+
extends TransactionalApplicationListenerSynchronization<AE>
99+
implements org.springframework.transaction.support.TransactionSynchronization {
100+
101+
public PlatformSynchronization(AE event, TransactionalApplicationListener<AE> listener,
102+
List<TransactionalApplicationListener.SynchronizationCallback> callbacks) {
103+
104+
super(event, listener, callbacks);
105+
}
106+
107+
@Override
108+
public void beforeCommit(boolean readOnly) {
109+
if (getTransactionPhase() == TransactionPhase.BEFORE_COMMIT) {
110+
processEventWithCallbacks();
111+
}
112+
}
113+
114+
@Override
115+
public void afterCompletion(int status) {
116+
TransactionPhase phase = getTransactionPhase();
117+
if (phase == TransactionPhase.AFTER_COMMIT && status == STATUS_COMMITTED) {
118+
processEventWithCallbacks();
119+
}
120+
else if (phase == TransactionPhase.AFTER_ROLLBACK && status == STATUS_ROLLED_BACK) {
121+
processEventWithCallbacks();
122+
}
123+
else if (phase == TransactionPhase.AFTER_COMPLETION) {
124+
processEventWithCallbacks();
125+
}
126+
}
127+
}
128+
129+
130+
private static class ReactiveSynchronization<AE extends ApplicationEvent>
131+
extends TransactionalApplicationListenerSynchronization<AE>
132+
implements org.springframework.transaction.reactive.TransactionSynchronization {
133+
134+
public ReactiveSynchronization(AE event, TransactionalApplicationListener<AE> listener,
135+
List<TransactionalApplicationListener.SynchronizationCallback> callbacks) {
136+
137+
super(event, listener, callbacks);
138+
}
139+
140+
@Override
141+
public Mono<Void> beforeCommit(boolean readOnly) {
142+
if (getTransactionPhase() == TransactionPhase.BEFORE_COMMIT) {
143+
return Mono.fromRunnable(this::processEventWithCallbacks);
144+
}
145+
return Mono.empty();
146+
}
147+
148+
@Override
149+
public Mono<Void> afterCompletion(int status) {
150+
TransactionPhase phase = getTransactionPhase();
151+
if (phase == TransactionPhase.AFTER_COMMIT && status == STATUS_COMMITTED) {
152+
return Mono.fromRunnable(this::processEventWithCallbacks);
153+
}
154+
else if (phase == TransactionPhase.AFTER_ROLLBACK && status == STATUS_ROLLED_BACK) {
155+
return Mono.fromRunnable(this::processEventWithCallbacks);
156+
}
157+
else if (phase == TransactionPhase.AFTER_COMPLETION) {
158+
return Mono.fromRunnable(this::processEventWithCallbacks);
159+
}
160+
return Mono.empty();
161+
}
162+
}
163+
89164
}

‎spring-tx/src/main/java/org/springframework/transaction/event/TransactionalEventListener.java

Lines changed: 8 additions & 7 deletions
Original file line numberDiff line numberDiff line change
@@ -1,5 +1,5 @@
11
/*
2-
* Copyright 2002-2021 the original author or authors.
2+
* Copyright 2002-2023 the original author or authors.
33
*
44
* Licensed under the Apache License, Version 2.0 (the "License");
55
* you may not use this file except in compliance with the License.
@@ -37,12 +37,13 @@
3737
* method allows you to prioritize that listener amongst other listeners running before
3838
* or after transaction completion.
3939
*
40-
* <p><b>NOTE: Transactional event listeners only work with thread-bound transactions
41-
* managed by a {@link org.springframework.transaction.PlatformTransactionManager
42-
* PlatformTransactionManager}.</b> A reactive transaction managed by a
43-
* {@link org.springframework.transaction.ReactiveTransactionManager ReactiveTransactionManager}
44-
* uses the Reactor context instead of thread-local variables, so from the perspective of
45-
* an event listener, there is no compatible active transaction that it can participate in.
40+
* <p>As of 6.1, transactional event listeners can work with thread-bound transactions managed
41+
* by a {@link org.springframework.transaction.PlatformTransactionManager} as well as reactive
42+
* transactions managed by a {@link org.springframework.transaction.ReactiveTransactionManager}.
43+
* For the former, listeners are guaranteed to see the current thread-bound transaction.
44+
* Since the latter uses the Reactor context instead of thread-local variables, the transaction
45+
* context needs to be included in the published event instance as the event source:
46+
* see {@link org.springframework.transaction.reactive.TransactionalEventPublisher}.
4647
*
4748
* <p><strong>WARNING:</strong> if the {@code TransactionPhase} is set to
4849
* {@link TransactionPhase#AFTER_COMMIT AFTER_COMMIT} (the default),

0 commit comments

Comments
 (0)