如何在 spring 反应器中将两个发布者合二为一

Question

我已经实现了一个虚拟反应存储库，但我正在努力使用更新方法：

@Override
public Mono<User> updateUser(int id, Mono<User> updateMono) {
    return  //todo with getUser
}

@Override
public Mono<User> getUser(int id) {
    return Mono.justOrEmpty(this.users.get(id));
}

一方面，我有新的发布者 Mono<User> updateMono，另一方面，我在 Mono.justOrEmpty(this.users.get(id)) 期间有另一个发布者。

如何将它们组合在一起，进行更新，并只回馈一个发布者？

我唯一想到的是：

@Override
public Mono<User> updateUser(int id, Mono<User> updateMono) {
    return  getUser(id).doOnNext(user -> {
        updateMono.subscribe(update -> {
            users.put(id, new User(id, update.getName(), update.getAge()));
            System.out.format("Updated user with id %d to %s%n", id, update);
        });
    });
}

是否正确？

Answer 1

见the reference guide on finding the right operator

值得注意的是，对于 Mono，您有 and、when、then（请注意，最后一个将在 3.1.0 中变为 flatMap，并且平面图将变为 flatMapMany)

doOnNext 更适用于日志记录或统计信息收集等辅助操作。 subscribe inside subscribe 是另一种糟糕的形式；通常你需要 flatMap 或类似的东西。

Answer 2

这几天玩了Spring 5个Reactive Streams功能，写了some sample codes（还没有public通过博客或twitter，我还需要更多的Reactor练习).

我也遇到了同样的问题，最后用一个Mono.zip更新了MongoDB中已有的item。

https://github.com/hantsy/spring-reactive-sample/blob/master/boot-routes/src/main/java/com/example/demo/DemoApplication.java

public Mono<ServerResponse> update(ServerRequest req) {

    return Mono
        .zip(
            (data) -> {
                Post p = (Post) data[0];
                Post p2 = (Post) data[1];
                p.setTitle(p2.getTitle());
                p.setContent(p2.getContent());
                return p;
            },
            this.posts.findById(req.pathVariable("id")),
            req.bodyToMono(Post.class)
        )
        .cast(Post.class)
        .flatMap(post -> this.posts.save(post))
        .flatMap(post -> ServerResponse.noContent().build());

}

更新：另一个用 Kotlin 编写的工作版本。

fun update(req: ServerRequest): Mono<ServerResponse> {
    return this.posts.findById(req.pathVariable("id"))
            .and(req.bodyToMono(Post::class.java))
            .map { it.t1.copy(title = it.t2.title, content = it.t2.content)    }
            .flatMap { this.posts.save(it) }
            .flatMap { noContent().build() }
}

如何在 spring 反应器中将两个发布者合二为一

how to combine two publisher in one in spring reactor

project-reactor

spring-webflux