The two `onto-chan` variants could be more flexible with the third argument
Description
As they stand, `onto-chan!` and `onto-chan!!` offer no help for inputs that need closing when exhausted. The typical example would be `line-seq`, but there is more to it (e.g. certain Streams also need closing). Generally speaking, and in-line with regular (non-async) Clojure where we have `with-open`, there should be an idiom to deal with such inputs in async Clojure. Here is at least one case where people have tripped over this.
The solution to this is pretty straight-forward - simply allow for the third argument to the two `onto-chan` variants to be a function (defaulting to `close!`). Something like the following:
Once this is in place one can trivially implement `line-chan` or `stream-chan`:
Is there any interest in this? I can provide a patch if need be...
Environment
Activity
The intent of onto-chan is to take coll and put it on a chan, so I don't think we're going to do this.
I think what you're asking for is more general async iterator handling, and that is something we're working on for both synchronous (in Clojure proper) and async (in core.async) variants. I don't think we've logged the async work already done on this, but the sync part is at https://clojure.atlassian.net/browse/CLJ-2555.
I am looking for a way to replicate the functionality of `with-open`, since lexical scoping doesn’t apply in async code. I followed and the attached patch, but frankly I don’t see how this idea helps me.
How would I implement line-chan (or stream-chan) today, while making sure that the underlying Reader (or Stream) is closed when exhausted (without essentially reinventing onto-chan)?
In any case, thanks for your prompt reply and feedback.
You won't want to use onto-chan! or go loops as they should not be used with (potentially) blocking io. I would spin a `thread` that sits in a loop/recur reading lines and feeding them to a channel. When the end is reached, close the stream and close the channel. I would probably not use line-seq at all - there's no reason to lazily seq in this case - just rely on the channel buffer backpressure to determine rate.
Assignee
Reporter
Labels
Approval
Patch
Priority
