This seems to have been introduced by stream chunking: #270

@fwbrasil, correct me if I am wrong, but I think we need a Chunk[V].map which produces Chunk[U < S] instead of Chunk[U] < S
https://github.com/getkyo/kyo/blob/main/kyo-core/shared/src/main/scala/kyo/chunks.scala#L84-L95

@hearnadam It doesn't specifically look related to chunking to me. Check out the top level definition of take:

The way I read that is, when invoked as take(0) that it still tries to process (runDiscard) the stream, just not passing it through any further downstream handlers (but anything upstream will still run to stream completion).

Yes, it doesn't seem related to chunking. I think we need a way for handlers to signal to producers that there is no interest in new elements anymore. It seems an easy way to do that is to resume with a Boolean instead of a Unit, indicating whether the producer should continue or not.

I've explored the solution I mentioned but it's not clear to me what the behavior should be with chunking. Does tap break chunking in ZIO?

ZStreams have operators that break chunking, as well as operators that expose the underlying chunk. Choosing an API here will depend on what's most important to end users. In my opinion, it's generally non-obvious that operators such as mapZIO break chunking...

https://github.com/zio/zio/blob/series/2.x/streams/shared/src/main/scala/zio/stream/ZStream.scala#L1850-L1853

To directly answer your question, yes .tap breaks chunking:

        for
            ref <- Ref.make(0)
            inc = ref.update(_   1)
            _ <- ZStream
                .range(0, 128, chunkSize = 8)
                .tap(_ => inc)
                .mapChunks { c =>
                    assert(c.size == 1)
                    c
                }
                .take(10)
                .runCollect
            count <- ref.get
        yield assert(count == 10)
        end for

@hearnadam While there is a decision here about if take(x) should break chunks / upstream effects should be limited to the number of elements taken, the main issue here is the upstream emits elements until it completes (rather than short circuiting early). The only difference the chunking decision would make is about the current chunk, not all following chunks. (Side note, I'm not a fan of the ZStream approach to chunking as a hidden concern and I hope kyo can avoid those limitations).

Thank you for the clarifications and links! If I understood correctly, we need to keep chunking but any handler in the chain of transformations should be able to halt execution for all its nested handlers, even they're still processing a chunk. This kind of "bi-directional" communication between nested handlers would be too expensive. It'd require double the effect suspensions.

Maybe we could have two separate effects:

1. The current Streams effect removing IOs/Fibers implementations. 100% pure effect, no chunking
2. Another effect adopting a pull-based approach using IOs/Fibers and chunking

I'm not sure what to name those but they seem both valuable independent directions to evolve the handling of this kind of computation

@lukestephenson @hearnadam would you have some feedback on my last comment? I can probably pick this up in the next weeks.

It seems an easy way to do that is to resume with a Boolean instead of a Unit, indicating whether the producer should continue or not.

This sounds like the best approach to me. Although I'd prefer to avoid the raw Boolean and have something like:

opaque type Ack = Boolean

object Ack {
  val Stop = false
  val Continue = true
}

If I understood correctly, we need to keep chunking but any handler in the chain of transformations should be able to halt execution for all its nested handlers, even they're still processing a chunk.

I don't care about the semantics within processing a chunk. If take indicated to terminate, I don't think that needs to cause termination of processing the in flight chunk, and agree this is too expensive. I don't want it to continue past the current chunk though.

I don't care about the semantics within processing a chunk. If take indicated to terminate, I don't think that needs to cause termination of processing the in flight chunk, and agree this is too expensive. I don't want it to continue past the current chunk though.

If this works, we could still keep a single Streams impl. I think we'd also need a mechanism to automatically break chunks if they're too large? What do you think @hearnadam?

fixed by #572