验证反应器流量的起始流

Question

flux
.bufferUntil(new Predicate<>() {
    private int count = 0;

    @Override
    public boolean test(T next) {
        return ++count >= N;
    }
})
// Zip with index to know the first element
.zipWith(Flux.<Integer, Integer>generate(() -> 0, (cur, s) -> {
    s.next(cur);
    return cur + 1;
}))
.map(t -> {
    if (t.getT2() == 0 && !validate(t.getT1()))
        throw new RuntimeException("Invalid");
    return t.getT1();
})
// Flatten buffered elements
.flatMapIterable(identity())

我本可以使用doonnext而不是第二个map，因为它不映射任何东西，但我不确定这是否是peek方法的可接受用法。

我也可以在第二个映射中使用一个有状态映射器来只运行一次，而不是用索引压缩，我想这是可以接受的，因为我已经使用了一个有状态谓词...

Answer 1

你的要求听起来很有趣！我们有switchonfirst，这对于验证第一个元素很有用。但是如果您有N个元素要验证，我们可以尝试这样的方法。

这里，我假设我必须验证前5个元素，它们应该<=5。则为有效流。否则，我们只会抛出错误，表示验证失败。

Flux<Integer> integerFlux = Flux.range(1, 10).delayElements(Duration.ofSeconds(1));

integerFlux
        .buffer(5)
        .switchOnFirst((signal, flux) -> {
            //first 5 elements are <= 5, then it is a valid stream
            return signal.get().stream().allMatch(i -> i <= 5) ? flux : Flux.error(new RuntimeException("validation failed"));
        })
        .flatMapIterable(Function.identity())
        .subscribe(System.out::println,
                System.out::println);   

然而，这种方法并不好，因为它每次都要收集5个元素，即使在第一次验证完成后也是如此，这可能是我们不想要的。

AtomicInteger atomicInteger = new AtomicInteger(1);
integerFlux
        .bufferUntil(i -> {
            if(atomicInteger.get() < 5){
                atomicInteger.incrementAndGet();
                return false;
            }
            return true;
        })
        .switchOnFirst((signal, flux) -> {
            return signal.get().stream().allMatch(i -> i <= 5) ? flux : Flux.error(new RuntimeException("validation failed"));
        })
        .flatMapIterable(Function.identity())
        .subscribe(System.out::println,
                   System.out::println);