等待多个语句使用R2DBC执行

Question

我需要用R2DBC执行多条语句，但是找不到有用的R2DBC示例.所以我的功能总是太早了：

public Publisher<Person> groupStatements(DatabaseClient client, Person person) {
// yes, I know that's harsh, but hey! so is JPA's @ElementCollection
return client.sql("DELETE FROM persons_address WHERE person = :id")
    .bind("id", person.getId())
    .fetch().rowsUpdated()
    .map(deleted -> {
        // now recreate every relationship
        GenericExecuteSpec statement = client.sql("INSERT INTO persons_address (person, address) VALUES (:person, :address)");
        person.getOfficePlaces().forEach(address -> {
            statement
                .bind("person", person.getId()).bind("address", address.getId())
                .fetch().rowsUpdated()  // there we go AWOL
                .subscribe(inserted -> {
                    // logging here
                });
        });
        return person;  //FIXME wait! need above grouped statements to complete
    });
}

注意:我使用H2作为后端。谢谢你提供的任何信息！

Answer 1

我找到了一种合适的批处理技术(这里替换了map/删除部分)，但是由于Statement#execute返回的发布者只有#订阅方法，所以我无法从链中返回。所以我给野兽喂食了几个齿轮

//DEBUG I couldn't figure out how to use labels! good enough
private static final String SQL_INSERT = "INSERT INTO persons_address (person, address) VALUES ($1, $2)";
...
.flatMap(deleted -> {
    if (person.getOfficePlaces().isEmpty()) {
        return Mono.just(person);
    } else {
        return client.inConnection(cnx -> {
            Statement stmt = cnx.createStatement(SQL_INSERT);
            person.getOfficePlaces().forEach(address -> {
                stmt.bind(0, person.getId()).bind(1, address.getId()).add();
            });
            return Flux.from(stmt.execute()).last().map(dontcare -> person);
        });
    }