bufferFrames

Buffers a frame flow without stranding the frames it is still holding.

Why this exists instead of buffer()

A Frame owns native memory and must be closed. The standard library's buffer(), flowOn, conflate and produceIn all move elements through a channel they create themselves, and none of them offers a hook for elements that never reach the collector. So a flow cancelled part way through, buffer().take(1) being the ordinary case, drops whatever is still queued on the floor: with the default capacity that is up to 64 undelivered frames, which for 1080p is hundreds of megabytes that nothing will ever free.

This operator owns its channel and closes every element the collector does not receive. Use it anywhere you would have reached for buffer(), and pass context anywhere you would have reached for flowOn: both hazards are the same hazard, and flowOn is the easier one to write by accident.

source.decodedFrames(video)
    .bufferFrames(context = Dispatchers.Default)   // decode off the collector's thread
    .take(10)                                      // the other 54 are closed, not leaked
    .collect { frame -> frame.use { render(it) } }

Frames that DO reach the collector are still the collector's to close, exactly as they are without this operator. This changes what happens to the ones that do not arrive, nothing else.

Parameters