Fork me on GitHub

Would like a small code review. Assume I want to listen to items coming out of a channel in and execute a blocking, side-effecty function my-fn! for each item. Is (pipeline-blocking 1 (chan (dropping-buffer 0)) my-fn! in) a good way to express this?


the reason you are using a dropping buffer there is because you want to throw away the results?


not having feedback from the output channel is going to effect the behavior of the pipeline


as a general principle, beyond core.async, functions returing useful values is a good thing, so you should try and do that, but it is not always possible. if you don't have a useful return value, I don't think pipeline is a good fit


if you don't have a useful return value, I might just spin up N threads that pull from the input and do whatever


if you look at pipeline blocking as a system with an input and an output and some capacity, the capacity being the number of items you can put in to the system, without removing items, before the system stops accepting new items


the capacity of pipeline-blocking is input buffer size + output buffer size + parallelism number


@hiredman: there are no interesting results in this case (it's at the edge of my system), so yeah might as well throw them away by (ab)using a dropping buffer. It just seemed to me that spinning up N processes would be reinventing (potentially poorly) a special case of pipeline-blocking


Definitely a noobie async question, but given a collection of channels, what's an idiomatic way to handle each of their taken values until they all close while determining how many are still open?


Merging and reducing implies waiting for them all to close before reading their values, alts! only takes the first, and all go-loop examples I've picked apart only handle taking from one channel repeatedly until it closes. I'm a bit stuck.


My use case is that I have n number of concurrent http requests and I want to update a progress bar, and that means knowing how many are closed and how many are still open.


there is a function (as opposed to a macro) version of alts that you can pass a collection of channels to


alt! is the macro, alts! is the function (had to remind myself)


so you combine that with a loop


alts! over all the channels, you get a result and a channel back, if the result is nil recur with the collection of channels minus that channel (because it is closed), if not do whatever with that result and recur with the same collection of channels


I would tend to avoid any kind of design that depends knowing how many channels are open vs closed, or does any kind of testing of channel open and closedness other than trying to read from a channel


Interesting! And also, interesting last comment there.


Realistically, the responses are coming from cljs-http so I know how many are being made.


I mean any kind of counting via, I dunno, filter or something, if you keep a counter in the loop, and assume all channels are open at the start and then decrement it when you get a nil


Ah okay, I think I get that. Thanks for explaining the difference between alt! and alts!. I overlooked that.


if you are counting the channels to limit the number of requests in flight you might want to checkout one of the pipeline variants


It's not so much a matter of limiting. I just want to update a progress bar (0%, 25%, 50%, 100%) for, say, 4 channels waiting to be closed.


I'm coming from the js / promise world which is taking some undoing in my head. 😉


ah, yeah, so you can just count the collection of channels, instead of worrying about open / closedness counts, because you will remove closed channels from the collection


I'll give that a shot! Thanks a lot @hiredman.


@hiredman when I run

(let [[v p] (alts! [c1 c2 c3 c4])]
          (println "GOT" v p))
I get GOT [nil many-to-many-channel] But when I look at some of the examples that are basically identical, v is the message taken from the channel, not nil. Am I missing something obvious?


Nil means the channel is closed, so you have a closed channel in there


And when I run it over and over, it produces a message.. sometimes.


The value is the result just like you used <!, The value if there is one, nil if the channel is closed


Ah, right, okay. Thanks for bearing with me.