| 6259 | return new StreamImpl(core.suspend(() => { |
| 6260 | const queue = new RingBuffer<A>(chunkSize) |
| 6261 | const emitOnStreamEnd = ( |
| 6262 | queueSize: number, |
| 6263 | channelEnd: Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown> |
| 6264 | ) => { |
| 6265 | if (queueSize < chunkSize) { |
| 6266 | const items = queue.toChunk() |
| 6267 | const result = Chunk.isEmpty(items) ? Chunk.empty<Chunk.Chunk<A>>() : Chunk.of(items) |
| 6268 | return pipe(core.write(result), core.flatMap(() => channelEnd)) |
| 6269 | } |
| 6270 | const lastEmitIndex = queueSize - (queueSize - chunkSize) % stepSize |
| 6271 | if (lastEmitIndex === queueSize) { |
| 6272 | return channelEnd |
| 6273 | } |
| 6274 | const leftovers = queueSize - (lastEmitIndex - chunkSize + stepSize) |
| 6275 | const lastItems = pipe(queue.toChunk(), Chunk.takeRight(leftovers)) |
| 6276 | const result = Chunk.isEmpty(lastItems) ? Chunk.empty<Chunk.Chunk<A>>() : Chunk.of(lastItems) |
| 6277 | return pipe(core.write(result), core.flatMap(() => channelEnd)) |
| 6278 | } |
| 6279 | const reader = ( |
| 6280 | queueSize: number |
| 6281 | ): Channel.Channel<Chunk.Chunk<Chunk.Chunk<A>>, Chunk.Chunk<A>, E, E, unknown, unknown> => |