RFR 8071597 Add Stream dropWhile and takeWhile operations

[Peter Levart]

On 06/03/2015 08:53 PM, Peter Levart wrote:
> Hi Paul,
>
> This is a usefull addition to Stream API for sequential ordered 
> streams. But does it have any utility in unordered streams at all? 
> Wouldn't it be better to just throw an IllegalStateException or 
> something if the stream is not ordered? I can't imagine currently a 
> situation where I would need to select an arbitrary sub-set of the 
> stream where all elements match a predicate and I would not be given 
> any guarantee about how big a subset I will get - the same with a 
> stream where an arbitrary sub-set of elements matching predicate is 
> taken away. If I wanted something similar for unordered stream, I 
> would rather say: "Give me at most N elements of the stream matching 
> predicate" for example - and that can already be achieved by 
> .filter().limit();
>
> So any use of [take|drop]While on unordered stream would most probably 
> be unintended and consequently could be considered a bug, don't you think?

Ok, I take back. I was only thinking in terms of the [take|drop]While 
Predicate being a function of the stream element. Now that I see an 
example where it is a function of time, I can understand it's utility 
even for unordered streams.

Regards, Peter

> Regards, Peter
>
> On 06/02/2015 03:13 PM, Paul Sandoz wrote:
>> Hi,
>>
>> Please review this webrev that adds take/dropWhile operations to streams:
>>
>>    http://cr.openjdk.java.net/~psandoz/jdk9/JDK-8071597-take-drop-while/webrev/
>>
>> I opted to weight the documentation of the operations towards ordered streams in the first paragraph. That is what makes most sense in terms of usage and what most people will read. Thus i refer to the "longest prefix" in the first paragraph then define what that means in subsequent paragraphs for ordered and unordered streams:
>>
>>   482     /**
>>   483      * Returns a stream consisting of the longest prefix of elements taken from
>>   484      * this stream that match the given predicate.
>>   485      *
>>   486      * <p>If this stream is ordered then the prefix is a contiguous sequence of
>>   487      * elements of this stream.  All elements of the sequence match the given
>>   488      * predicate, the first element of the sequence is the first element
>>   489      * (if any) of this stream, and the element (if any) immediately following
>>   490      * the last element of the sequence does not match the given predicate.
>>   491      *
>>   492      * <p>If this stream is unordered then the prefix is a subset of elements of
>>   493      * this stream.  All elements (if any) of the subset match the given
>>   494      * predicate.  In this case the behavior of this operation is
>>   495      * nondeterministic; it is free to select any valid subset as the prefix.
>>   496      *
>>   497      * <p>This is a <a href="package-summary.html#StreamOps">short-circuiting
>>   498      * stateful intermediate operation</a>.
>>   499      *
>> ...
>>   528     default Stream<T> takeWhile(Predicate<? super T> predicate) {
>>
>>   537     /**
>>   538      * Returns a stream consisting of the remaining elements of this stream
>>   539      * after dropping the longest prefix of elements that match the given
>>   540      * predicate.
>>   541      *
>>   542      * <p>If this stream is ordered then the prefix is a contiguous sequence of
>>   543      * elements of this stream.  All elements of the sequence match the given
>>   544      * predicate, the first element of the sequence is the first element
>>   545      * (if any) of this stream, and the element (if any) immediately following
>>   546      * the last element of the sequence does not match the given predicate.
>>   547      *
>>   548      * <p>If this stream is unordered then the prefix is a subset of elements of
>>   549      * this stream.  All elements (if any) of the subset match the given
>>   550      * predicate.  In this case the behavior of this operation is
>>   551      * nondeterministic; it is free to select any valid subset as the prefix.
>>   552      *
>> ...
>>   584     default Stream<T> dropWhile(Predicate<? super T> predicate) {
>>
>>
>> After this has been reviewed i will follow up with a further issue regarding the specification of takeWhile, stateful predicates and cancellation. I avoided such specification here as it's likely to rathole :-)
>>
>> Basically the takeWhile operation is implemented such that one can do:
>>
>>       long t = System.currentTimeMillis();
>>       List<BigInteger> pps = Stream
>>           .generate(() -> BigInteger.probablePrime(1024, ThreadLocalRandom.current()))
>>           .parallel()
>>           .takeWhile(e -> (System.currentTimeMillis() - t) < TimeUnit.SECONDS.toMillis(5))
>>           .collect(toList());
>>
>> Paul.
>