reactive-programming - 抛出异常作为 react 流验证的最正确方法

Question

我有一个 react 流，我想要一个步骤来应用验证检查，如果失败，将引发异常。是否有一种普遍接受的风格可以做到这一点？据我所知，我在 Mono 中有三个选项(使用 then() ) , filter() , 和 map() .

filter()最接近我想要的流，因为我实际上并没有尝试更改流中的数据类型或切换到另一个流。但是，filter 应该返回 true/false 以过滤掉项目，所以总是返回 TRUE 有点愚蠢。

then()让我专门选择错误/成功排放，但有时对于这种类型的验证，我无法轻松地将其拆分为它自己的私有(private)方法，并且样板文件使流声明更难以阅读。

map()与使用 filter() 几乎相同, 除非你总是在输入中返回而不是 TRUE。

作为一个非常人为的例子，考虑一个服务，它有一个包含 0 个或多个字母的列表来发送给一个人:

public interface Person {
    UUID getId();
    List<String> getKnownLanguages();
}

public interface Letter {
    String getLanguage();
}

public class LetterService {
    private Letter findOneLetterForPerson(final UUID id) { /* ... */ }

    private void removeLetter(final Letter letter) { /* ... */ }
}

创建如下所示的方法的更好选择是什么:

public Mono<Optional<Letter>> getNextValidLetterForPerson(final Person person) {
    return Mono.just(person)
               .and(this::getNextLetterForPerson)

               /////////////////////////////////////////
               //
               .filter(this::validatePersonCanReadLetter1)
               .map(Tuple2::getT2)
               //
               // OR
               //
               .then(this::validatePersonCanReadLetter2)
               //
               // OR
               //
               .map(this::validatePersonCanReadLetter3)
               //
               /////////////////////////////////////////

               // If the letter was invalid for the person, remove the letter from the
               // the system as a side effect, and retry retrieving a letter to send
               .doOnError(this::removeInvalidLetter)
               .retry(this::ifLetterValidationFailed)

               // Map the result to an appropriate Optional
               .map(Optional::of)
               .defaultIfEmpty(Optional.empty());
}

上例中使用的支持方法有:

public static class LetterInvalidException extends RuntimeException {
    private Letter mLetter;
    public LetterInvalidException(final Letter letter) { mLetter = letter; }
    public Letter getLetter() { return mLetter; }
}


/** Gets the next letter for a person, as a reactive stream */
private Mono<Letter> getNextLetterForPerson(final Person person) {
    return Mono.create(emitter -> {
        final Letter letter = mLetterService.findOneLetterForPerson(person.getId());

        if (letter != null) {
            emitter.success(letter);
        }
        else {
            emitter.success();
        }
    });
}

/** Used to check whether the cause of an error was due to an invalid letter */
private boolean ifLetterValidationFailed(final Throwable e) {
    return e instanceof LetterInvalidException;
}

/** Used to remove an invalid letter from the system */
private void removeInvalidLetter(final Throwable e) {
    if (ifLetterValidationFailed(e)) {
        mLetterService.removeLetter(((LetterInvalidException)e).getLetter());
    }
}

/*************************************************************************
 *
 *************************************************************************/

private boolean validatePersonCanReadLetter1(final Tuple2<Person, Letter> tuple) {
    final Person person = tuple.getT1();
    final Letter letter = tuple.getT2();

    if (!person.getKnownLanguages().contains(letter.getLanguage())) {
        throw new LetterInvalidException(letter);
    }

    return true;
}

private Mono<Letter> validatePersonCanReadLetter2(final Tuple2<Person, Letter> tuple) {
    return Mono.create(emitter -> {
        final Person person = tuple.getT1();
        final Letter letter = tuple.getT2();

        if (!person.getKnownLanguages().contains(letter.getLanguage())) {
            emitter.error(new LetterInvalidException(letter));
        }
        else {
            emitter.success(letter);
        }

    });
}

private Letter validatePersonCanReadLetter3(final Tuple2<Person, Letter> tuple) {
    final Person person = tuple.getT1();
    final Letter letter = tuple.getT2();

    if (!person.getKnownLanguages().contains(letter.getLanguage())) {
        throw new LetterInvalidException(letter);
    }

    return letter;
}

理想情况下，我会喜欢一种方法，例如 Mono<T> validate(..)这将允许测试流项目并返回或抛出异常(如果返回，框架会将其视为错误)，但我对响应式(Reactive)编程相当陌生，没有看到任何类似的工作。

Answer 1

也许 handle是一个更好的解决方案，它可以作为 map 和过滤器的组合:

Mono.just(p).and(test::getNextLetterForPerson).handle((tuple, sink) -> {
    final Person person = tuple.getT1();
    final Letter letter = tuple.getT2();

    if (!person.getKnownLanguages().contains(letter.getLanguage())) {
        sink.error(new LetterInvalidException(letter));
        return;
    }

    sink.next(letter);
}).subscribe(value -> System.out.println(((Letter) value).getLanguage()),
t -> System.out.println(t.getMessage()));

如您所见，它几乎就像您的 validatePersonCanReadLetter3