takeUntil? Đang xử lý dữ liệu đầu vào được chunked. Làm sao?

Question

Giả sử tôi có một máy phát điện, mà trả về cho tôi một số kết quả chia thành khối, mà tôi muốn kéo vào một danh sách phẳng:

def pull(chunk: Chunk, result: Stream[Item] = Stream.empty): Stream[Item] = { 
    val soFar = chunk.items ++ result 
    if(chunk.hasNext) pull(generator.next(chunk), soFar) else soFar 
} 

Về mặt lý thuyết, đây là những gì tôi muốn, ngoại trừ, rồi nó lấy toàn bộ nội dung trả trước, và tôi muốn nó lười biếng. Một cái gì đó như thế này:

Stream.iterate(generator.first)(generator.next) 
    .takeWhile(_.hasNext) 
    .flatMap(_.items) 

hầu hết hoạt động, nhưng loại bỏ đoạn cuối cùng.

Có vẻ như tôi cần một số .takeUntil tại đây: như takeWhile, nhưng đi qua toàn bộ chuỗi trước khi chấm dứt. Làm thế nào để tôi làm điều này một cách tự nhiên?

Answer 1

Đây là những gì tôi đã đưa ra ... Trông kinda yucky nhưng điều này là tốt nhất tôi có thể nghĩ ra:

Stream.iterate(generator.first) { 
    case chunk if chunk.hasNext => generator.next 
    case _ => null 
}.takeWhile(_ != null) 
    .flatMap(_.items) 

Bất kỳ ý tưởng tốt hơn?