usingWhen factory issue with reactor-core 3.4.17

[srikanthpolineni]

Hands-On-Reactive-Programming-in-Spring-5/chapter-04/src/test/java/org/rpis5/chapters/chapter_04/ReactorEssentialsTest.java

Line 420 in 915dd98

Flux.usingWhen(

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]

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.