New method for a completeable `takeWhile`

[JustinPihony]

Short description

Create a takeWhile method which handles uncommitted items.

Details

Right now if you use a takeWhile then it will shut down the stream as soon as the predicate fails. If there is anything that is not committed but past the takeWhile then it will not be completed/drained. It would be good to create a new mechanism to stop taking new items, yet not fully stop the stream until the existing processing is completed. Maybe a startTakeWhile and completeTakeWhile - or something along those lines.