用于 Streams 的包含 takeWhile()

Question

我想知道是否有办法添加针对方法 takeWhile() 的条件测试的流的最后一个元素。我相信我想实现类似于 RxJava 的 takeUntil() 方法的东西。

我猜没有直接的方法可以做到这一点（如果我错了，请纠正我），但我想知道是否有适当的解决方法来实现这一点，我现在知道了。

我在 Stackoverflow 中进行了搜索，但几乎没有成功。如果您认为有可以解决我的问题的线程，我当然希望看到它。

如果您查看以下代码的 peek()，您将看到数字 5 已根据 takeWhile() 条件进行检查，但它从未到达 forEach() 中：

IntStream.of(1, 3, 2, 5, 4, 6)
        .peek(foo -> System.out.println("Peek: " + foo))
        .takeWhile(n -> n < 5)
        .forEach(bar -> System.out.println("forEach: " + bar));

预期结果是根据 takeWhile 的条件检查的最后一个元素到达 forEach 的 System.out::println。在这种情况下，它是 5。

谢谢大家！

Answer 1

另一种方法是使用有状态谓词 - 如果您尝试并行化流，这将是不可预测的，但只要您不需要这样做：

private static <T> Predicate<T> inclusiveFirstFailed(final Predicate<T> p) {
  final var goOn = new AtomicBoolean(true);
  return t -> p.test(t) ? goOn.get() : goOn.getAndSet(false);
}

在您的情况下，您需要使用 IntPredicate：

private static IntPredicate inclusiveFirstFailed(final IntPredicate p) {
  final var goOn = new AtomicBoolean(true);
  return t -> p.test(t) ? goOn.get() : goOn.getAndSet(false);
}

使用示例：

IntStream.of(1, 3, 2, 5, 4, 6)
        .peek(foo -> System.out.println("Peek: " + foo))
        .takeWhile(inclusiveFirstFailed(n -> n < 5))
        .forEach(bar -> System.out.println("forEach: " + bar));

Answer 2

使用普通的流 API 没有方便的方法来做到这一点。这是可能的，以一种丑陋的方式（您需要调整这个实现，这只是takeWhileJava 8 的正常“向后移植”）。

这家伙编写了一个流扩展库，其中包含takeWhileInclusive.

使用示例：

IntStreamEx.of(1, 3, 2, 5, 4, 6)
    .peek(foo -> System.out.println("Peek: " + foo))
    .takeWhileInclusive(n -> n < 5)
    .forEach(bar -> System.out.println("forEach: " + bar));