JdbcReactiveSqlExecutor extends JdbcSqlExecutor implements ReactiveSqlExecutor { @Override public Mono<Void> execute(Publisher<SqlRequest> request) { return Mono.defer(() -> toFlux(request) .flatMap(sqlRequest -> getConnection(sqlRequest) .flatMap(connection -> Mono.<Void>fromSupplier(() -> { doExecute(connection, sqlRequest); return null; }).doFinally(type -> releaseConnection(connection, sqlRequest)))).then()); } JdbcReactiveSqlExecutor(); abstract Mono<Connection> getConnection(SqlRequest sqlRequest); abstract void releaseConnection(Connection connection, SqlRequest sqlRequest); @Override Mono<Integer> update(Publisher<SqlRequest> request); @Override Mono<Void> execute(Publisher<SqlRequest> request); @Override Flux<E> select(Publisher<SqlRequest> request, ResultWrapper<E, ?> wrapper);  }
@Test public void testExecute() { Mono<Void> ddl = sqlExecutor.execute(Mono.just(of("create table test( id varchar(32) primary key )"))) .doOnError(Throwable::printStackTrace); StepVerifier.create(ddl) .verifyComplete(); Mono<Integer> counter = sqlExecutor .update(Flux.range(0, 10) .doOnNext(i -> System.out.println()) .map(num -> prepare("insert into test (id) values (?) ", num))) .doOnError(Throwable::printStackTrace); StepVerifier.create(counter) .expectNext(10) .verifyComplete(); Mono<Long> data = sqlExecutor.select(Mono.just(of("select * from test")), map()) .doOnError(Throwable::printStackTrace) .map(map -> map.get("ID")) .count(); StepVerifier.create(data) .expectNext(10L) .verifyComplete(); Mono<Long> count = sqlExecutor.select(Flux.range(0, 10) .map(String::valueOf) .map(num -> of("select * from test where id = ?", num)), map()) .doOnError(Throwable::printStackTrace) .map(map -> map.get("ID")) .count(); StepVerifier.create(count) .expectNext(10L) .verifyComplete(); }