usingWhen factory issue with reactor-core 3.4.17
srikanthpolineni opened this issue · 1 comments
srikanthpolineni commented
UsingWhen factory method declaration got changed in reactor-core 3.4.17. Source needs to update with
@Test
public void usingWhenExample() throws InterruptedException {
Flux.usingWhen(
Transaction.beginTransaction(),
transaction -> transaction.insertRows(Flux.just("A", "B")),
Transaction::commit,
(transaction, throwable) -> Mono.empty(), // <<== extra parameter
Transaction::rollback
).subscribe(
d -> log.info("onNext: {}", d),
e -> log.info("onError: {}", e.getMessage()),
() -> log.info("onComplete")
);
Thread.sleep(1000);
}
Linukso1D commented
Hi there I would say your extra parameter for the asyncError was wrong because it should perform the rollback if an exception occurs at transaction.insertRows
From the other side if an error occurs at commit - rollback will not be called and will be passed to a subscriber
Flux.usingWhen(
Transaction.beginTransaction(),
transaction -> transaction.insertRows(Flux.just("A", "B")),
transaction -> transaction.commit().onErrorResume(ex -> transaction.rollback().then(Mono.error(ex))), // this approach is able to catch an error inside the commit block. PS
(transaction, throwable) -> transaction.rollback(), // rollback on error
Transaction::rollback
)
I would be happy to look at some shorter forms for tx::commit
but for now I don't know how to catch an error for commit
block in some pretty way.