New issue
Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.
By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.
Already on GitHub? Sign in to your account
Omit cancellation of transactional Monos in TransactionOperator #23562
Changes from all commits
File filter
Filter by extension
Conversations
Jump to
Diff view
Diff view
There are no files selected for viewing
Original file line number | Diff line number | Diff line change |
---|---|---|
|
@@ -71,6 +71,21 @@ public ReactiveTransactionManager getTransactionManager() { | |
return this.transactionManager; | ||
} | ||
|
||
@Override | ||
public <T> Mono<T> transactional(Mono<T> mono) { | ||
return TransactionContextManager.currentContext().flatMap(context -> { | ||
Mono<ReactiveTransaction> status = this.transactionManager.getReactiveTransaction(this.transactionDefinition); | ||
// This is an around advice: Invoke the next interceptor in the chain. | ||
// This will normally result in a target object being invoked. | ||
// Need re-wrapping of ReactiveTransaction until we get hold of the exception | ||
// through usingWhen. | ||
return status.flatMap(it -> Mono.usingWhen(Mono.just(it), ignore -> mono, | ||
this.transactionManager::commit, s -> Mono.empty()) | ||
There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. note that in case of cancellation, since no There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. Commit on cancel is what we want if just a subset of data is processed. In imperative code this compares to processing a subset of a result list. The error case is ignored because errors are handled after the There was a problem hiding this comment. Choose a reason for hiding this commentThe reason will be displayed to describe this comment to others. Learn more. makes sense, thanks for the explanations. nothing to add 👍 |
||
.onErrorResume(ex -> rollbackOnException(it, ex).then(Mono.error(ex)))); | ||
}) | ||
.subscriberContext(TransactionContextManager.getOrCreateContext()) | ||
.subscriberContext(TransactionContextManager.getOrCreateContextHolder()); | ||
} | ||
|
||
@Override | ||
public <T> Flux<T> execute(TransactionCallback<T> action) throws TransactionException { | ||
|
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
wouldn't it be better to start from
Mono.usingWhen
?There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
We require the status value outside of
usingWhen(…)
(seerollbackOnException
)