Inclusive takeWhile() for Streams

Question

I want to know if there is a way to add the last element of the stream that was tested against the condition of the method takeWhile(). I believe I want to achieve something similar to RxJava's takeUntil() method.

I'm guessing there is no direct way to do this (correct me if I am mistaken), but I would like to know if there is a proper workaround to achieve this that I am now aware of.

I've searched throughout Stack Overflow with little to no success. If you think there are threads that could solve my problems, I would certainly like to see it.

If you look at the following code's peek() you will see that the number 5 is checked against the takeWhile() condition but it never arrives in the 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));

The expected result is for the last element checked against takeWhile's condition to arrive at the forEach's System.out::println. In this case it is 5.

Thanks to everyone!

Answer 1

There's no convenient way to do that with the normal stream API. It is possible in an ugly way (you would need to adapt this implementation, that's just a normal takeWhile "backported" for Java 8).

This guy has written a stream extension library which has takeWhileInclusive.

Sample usage:

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));

Answer 2

An alternative would be to use a stateful predicate - this would be unpredictable if you try to parallellize the stream but as long as you do not need to do that:

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);
}

In your case you would need to use an IntPredicate:

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

Sample usage:

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 3

I add an horrible approach that might be overkill for your (or any) use case, but just for fun.

public static void main(String[] args) {
    List<Integer> source = List.of(1, 3, 2, 5, 4, 6);
    Iterator<Integer> iterator = source.iterator();
    AtomicBoolean proceed = new AtomicBoolean(true);

    Stream
        .generate(() -> {
            if (!proceed.get() || !iterator.hasNext()) {
                return null;
            }
            int value = iterator.next();
            System.out.println("generate: " + value);
            proceed.set(value < 5);
            return value;
        })
        .takeWhile(Objects::nonNull)
        .forEach(bar -> System.out.println("forEach: " + bar));

}

The output will be:

generate: 1
forEach: 1
generate: 3
forEach: 3
generate: 2
forEach: 2
generate: 5
forEach: 5

Probably the worst thing about this approach is that it gives generate() a responsibility (checking if there are more) that it does not belong with it.

Answer 4

Another way for linear Streams would be the iterative approach:

public static <T> Stream<T> takeWhileIncl(Stream<T> stream, Predicate<T> pred) {
    var res = new ArrayList<T>();
    for (var e : stream.toList()) {
        res.add(e);
        if (pred.test(e))
            continue;
        break;
    }
    return res.stream();
}

Answer 5

The limitUntilClosed method from jOOl does exactly what you need.

For reference: https://www.jooq.org/products/jOO%CE%BB/javadoc/latest/org.jooq.jool/org/jooq/lambda/Seq.html#limitUntilClosed(java.util.function.Predicate)