reactive-programming - 使用项目reactor mergeWith()运算符来实现 "if/elseif/else"分支逻辑

Question

我正在尝试使用项目 react 器 mergeWith运算符以实现 if/elseif/else分支逻辑如下所述:RxJS, where is the If-Else Operator .

提供的示例是用 RxJS 编写的，但基本思想保持不变。

基本上这个想法是使用 filter 3 上的运算符 monos/publishers (因此有 3 个不同的谓词)并合并 3 monos如下(这里当然是RxJS Observables):

const somethings$ = source$
  .filter(isSomething)
  .do(something);

const betterThings$ = source$
  .filter(isBetterThings)
  .do(betterThings);

const defaultThings$ = source$
  .filter((val) => !isSomething(val) && !isBetterThings(val))
  .do(defaultThing);

// merge them together
const onlyTheRightThings$ = somethings$
  .merge(
    betterThings$,
    defaultThings$,
  )
  .do(correctThings);

我已复制并粘贴上述文章中的相关示例。

考虑something$ , betterThings$和defaultThings$我们的单声道是isSomething & isBetterThings是谓词。

现在这是我的 3 个真实的 monos/publishers (用java编写):

private Mono<ServerResponse> validateUser(User user) {
    return Mono.just(new BeanPropertyBindingResult(user, User.class.getName()))
        .doOnNext(err -> userValidator.validate(user, err))
        .filter(AbstractBindingResult::hasErrors)
        .flatMap(err ->
            status(BAD_REQUEST)
                .contentType(APPLICATION_JSON)
                .body(BodyInserters.fromObject(err.getAllErrors()))
        );
}

private Mono<ServerResponse> validateEmailNotExists(User user) {
    return userRepository.findByEmail(user.getEmail())
        .flatMap(existingUser ->
            status(BAD_REQUEST)
                .contentType(APPLICATION_JSON)
                .body(BodyInserters.fromObject("User already exists."))
        );
}

private Mono<ServerResponse> saveUser(User user) {
    return userRepository.save(user)
        .flatMap(newUser -> status(CREATED)
            .contentType(APPLICATION_JSON)
            .body(BodyInserters.fromObject(newUser))
        );
}

这是需要合并三个 publishers 的顶级方法:

public Mono<ServerResponse> signUpUser(ServerRequest serverRequest) {
    return serverRequest.bodyToMono(User.class)
        .mergeWith(...)

}

我不知道如何使用 mergeWith()运算符...我已经尝试过 Mono.when()静态运算符需要多个发布者(对我来说很好)但返回 Mono<void> (对我来说不好)。

有人可以帮忙吗？

P.S.我相信您会原谅 RxJS (js) 和 Reactor 代码 (java) 之间的混合。我打算利用 RxJS 中的知识在我的 Reactor 应用程序中实现类似的目标。 :-)

编辑1:我已经尝试过:

public Mono<ServerResponse> signUpUser(ServerRequest serverRequest) {
    return serverRequest
        .bodyToMono(User.class)
        .flatMap(user -> validateUser(user).or(validateEmailNotExists(user)).or(saveUser(user))).single();
}

但我收到此错误:NoSuchElementException: Source was empty

编辑 2:与以下相同(注意括号):

public Mono<ServerResponse> signUpUser(ServerRequest serverRequest) {
    return serverRequest
        .bodyToMono(User.class)
        .flatMap(user -> validateUser(user).or(validateEmailNotExists(user)).or(saveUser(user)).single());
}

编辑 3:与 Mono<User> 相同的错误:

public Mono<ServerResponse> signUpUser(ServerRequest serverRequest) {
    Mono<User> userMono = serverRequest.bodyToMono(User.class);
    return validateUser(userMono)
        .or(validateEmailNotExists(userMono))
        .or(saveUser(userMono))
        .single();
}

编辑4:我可以确认三个单声道中至少有一个会始终发射。这是当我使用or()时运算符(operator)发现出了问题...

如果我使用它，我的所有测试都会通过:

public Mono<ServerResponse> signUpUser(ServerRequest serverRequest) {
    return serverRequest.bodyToMono(User.class)
        .flatMap(user -> Flux.concat(validateUser(user), validateEmailNotExists(user), saveUser(user)).next().single());
}

我用过concat()此处的运算符保留运算顺序。

你知道我对 or() 的理解有什么问题吗？运算符？

编辑 5:我已尝试使用 cache()运算符如下无济于事:

public Mono<ServerResponse> signUpUser(ServerRequest serverRequest) {
    return serverRequest
        .bodyToMono(User.class)
        .cache()
        .flatMap(user -> validateUser(user)
            .or(validateEmailNotExists(user))
            .or(saveUser(user))
            .single()
        );
}

Answer 1

您当前的代码示例意味着您的 3 个方法返回 Mono<ServerResponse>应该采取Mono<User>而不是User ，因此您可能需要在那里进行一些更改。

但是，我离题了 - 这似乎不是这里的主要问题。

根据我对该链接中描述的模式的理解，您正在创建 3 个单独的 Mono对象，其中只有一个会返回结果 - 并且您需要一个 Mono无论您的原始 3 个 Mono 中的哪一个对象返回。

在这种情况下，我会推荐如下内容:

Mono<ServerResult> result = Flux.merge(validateUser(user), validateEmailNotExists(user), saveUser(user)).next().single();

分解:

• 静态Flux.merge()方法需要你的 3 Mono对象并将它们合并为 Flux ;
• next()将第一个可用结果返回为 Mono ;
• single()将确保Mono发出一个值，而不是什么都不发出，否则抛出异常。 (可选，但只是一点安全网。)

您也可以直接链接 Mono.or()像这样:

Mono<ServerResult> result = validateUser(user).or(validateEmailNotExists(user)).or(saveUser(user)).single();

这种方法的优点是:

• 在某些情况下可以说它更具可读性；
• 如果您可能有多个 Mono在您的链中返回一个结果，这允许您设置选择的优先顺序(与上面的示例相反，您将首先获得 Mono 发出的值。)

潜在的缺点之一是性能。如果saveUser()上面代码中先返回一个值，然后还得等待另外两个Mono在合并之前要完成的对象Mono将完成。