Fix Enum.take for Stream.chunk_by

[igas]

While reviewing #3016 I found a bug in chunk_by:

stream = Stream.chunk_by([1, 2, 2, 3, 4, 4, 6, 7, 7], &(rem(&1, 2) == 1))
stream |> Enum.take(2)
# => [[1], [2, 2], [3]]

[josevalim]

Please avoid changing style of other assertions as it makes it hard to see what is being changed as part of this change. :)

[igas]

Sorry about that, I just went through each test and add take assertion where it missed, to prevent such things in future. Below I added line 85.

[josevalim]

Good catch!

[josevalim]

Unfortunately this fix is not correct. We cannot access the accumulator state because the take value can be nested inside other states or it could just mean whatever. :)

[josevalim]

Thanks for finding this out! I have pushed a fix to Enum.take/2 as it was continue to emit items when counter reached 0!

[igas]

👍