Mono.when not respond

[kevin70]

versions:

java 11
r2dbc-spi: 0.9.0-RELEASE
r2dbc-mysql: 0.9.0-SNAPSHOT

code:

Mono<Void> f1(Connection connection) {
  return Flux.from(connection.createStatement("this is bad sql").execute())
      .flatMap(Result::getRowsUpdated)
      .then();
}

Mono<Void> f2(Connection connection) {
  return Flux.from(connection.createStatement("this is bad sql").execute())
      .flatMap(Result::getRowsUpdated)
      .then();
}

Mono<Void> execWhen(Connection connection) {
  return Mono.when(f1(connection), f2(connection));
}

@Test
void testFailed() {
  var connectionFactory = super.rc.getConnectionFactory();
  Flux.usingWhen(
          connectionFactory.create(),
          connection -> {
            // transaction
            return Flux.from(connection.beginTransaction())
                .thenMany(execWhen(connection))
                .concatWith(connection.commitTransaction())
                .onErrorResume(
                    t -> Mono.from(connection.rollbackTransaction()).then(Mono.error(t)));
          },
          Connection::close)
      .as(StepVerifier::create)
      .expectError(R2dbcBadGrammarException.class)
      .verify(Duration.ofSeconds(10));
}

results:

AssertionError: VerifySubscriber timed out on reactor.core.publisher.FluxUsingWhen$ResourceSubscriber

expect:

R2dbcBadGrammarException

When I use Mono.when when an exception occurs when connecting multiple publishers, the upstream will not return the correct result.

[kevin70]

Mono<Void> f1(Connection connection) {
  return Flux.from(connection.createStatement("this is bad sql").execute())
      .flatMap(Result::getRowsUpdated)
      .then();
}
Mono<Void> f2(Connection connection) {
  return Flux.from(connection.createStatement("this is bad sql").execute())
      .flatMap(Result::getRowsUpdated)
      .then();
}
Mono<Void> execThen(Connection connection) {
  return f1(connection).then(f2(connection));
}
@Test
void testSuccess() {
  var connectionFactory = super.rc.getConnectionFactory();
  Flux.usingWhen(
          connectionFactory.create(),
          connection -> {
            // transaction
            return Flux.from(connection.beginTransaction())
                .thenMany(execThen(connection))
                .concatWith(connection.commitTransaction())
                .onErrorResume(
                    t -> Mono.from(connection.rollbackTransaction()).then(Mono.error(t)));
          },
          Connection::close)
      .as(StepVerifier::create)
      .expectError(R2dbcBadGrammarException.class)
      .verify(Duration.ofSeconds(10));
}

I modify the code to get the correct result in the above case.