reactor/reactor

Hello!

I try to insert a list of items to PostgreSQL database via R2DBC, but it does not complete. Configurations with example are below.

Could anyone get an advise why this code does not hit a database?

    private Flux<OnlineChartItem> batchInsert() {
        return Mono.from(chartInputConnectionFactory.create())
                .log("start#1")
                .flatMapMany(connection -> {
                    Statement statement = connection.createStatement(INSERT_INTO_ONLINE_CHARTS);
                    for (OnlineChartItem item : items) {
                        this.prepareInsertStatement(statement, item).add();
                    }
                    return Flux.from(statement.execute());
                })
                .thenMany(Flux.fromIterable(items));
    }

    @Bean
    public ConnectionFactory chartInputConnectionFactory(R2dbcProperties r2dbcProperties) {
        ConnectionFactoryOptions factoryOptions = ConnectionFactoryOptions.parse("r2dbc:pool:postgresql://localhost:5432/my_db")
                .mutate()
                .option(ConnectionFactoryOptions.USER, r2dbcProperties.getUsername())
                .option(ConnectionFactoryOptions.PASSWORD, r2dbcProperties.getPassword())
                .build();
        ConnectionFactory connectionFactory = ConnectionFactories.get(factoryOptions);

        R2dbcProperties.Pool pool = r2dbcProperties.getPool();
        ConnectionPoolConfiguration configuration = ConnectionPoolConfiguration.builder()
                .connectionFactory(connectionFactory)
                .maxSize(20)
                .initialSize(10)
                .acquireRetry(Integer.MAX_VALUE)
                .maxAcquireTime(Duration.ofSeconds(2))
                .validationQuery("select 1")
                .build();
        return new ConnectionPool(configuration);
    }