This next bit is NOT KOTLIN (or at least it's only Kotlin in that I couldn't do it using Sequence!) I started with

generateSequence

, but

Sequence

doesn't implement `Closeable`/`AutoCloseable` which makes it hard to hand it off to another object and let that object clean up, so I switched to this:

java.util.stream.Stream
  .generate { /* work to parse out a single object from the inputStream */ }
  .takeWhile { it != null }
  .onClose { inputStream.close() }

However, I now can't use

stream.parallel().forEach

, because it ends up trying to process the actual JSON tokens in parallel rather than the resulting objects. So my theory is that I need to read the objects in in batches of I don't know, 16 or so, and then process those 16 in parallel, read in the next batch, and so on. I'd like the client using it to still see it as a single

Stream<Thing>

though - it shouldn't need to care about the underlying implementation...

Is there something better I don't know about? Perhaps in the coroutines libs?

for coroutines, there’s

awaitAll

to wait for a list of async operations to complete

but with Java Streams, there is a

sequential()

method

oh wait generate is sequential by default

I think your issue is with the consumer; you meant to use

.forEachOrdered

I'll have a look at

Flow

, see if it might have some lib methods to save me trying to do it myself

but with flow, you can definitely

flow {
    file.bufferedReader().use { while (true) emit(it.readlnOrNull() ?: break) }
}

and the reader will get closed whether the flow is fully consumed or the collector is cancelled