Morning

I have a core.async design question

I have a periodic job that runs to get the status of a set of package shipments from various couriers (dhl, fedex, etc)

so I create a set of channels, issue the api call and put the returned result on the channel

I then core.async/merge those channels and then use alts! in a go-loop to collect all the results

Here's my wait function

(defn wait-values [wait channels]
  (let [all (merge channels)
        t-out (timeout wait)]
    (go-loop [values []]
      (let [[value _] (alts! [all t-out])]
        (if (nil? value)
          values
          (recur (conj values value)))))))

Now this is where my understanding of core.async perhaps isn't as good as it needs to be: how would this scale?

It's probably fine if I have a small number of shipments, but what about when the number of requests and channels gets large when there are 100's or 1000's of shipments to be tracked?

why do you merge the channels?

I would have thought that you'd do the alts on all the channels

also, I do think that Timothy Baldrigde has a demo in here https://www.youtube.com/watch?v=enwIIGzhahw where he shows usage of 1000 channels

I guess at around 30 minutes into the talk.

nah, he doesn't alts on that many channels.

hmm, merge is redundant?

I can just alts! on (conj channels t-out)

@matttylr: if you want to introduce parallelism you may consider using a pipeline instead of a single go-loop

hmmm, maybe not since you just want to get values until a timeout fires

I had looked briefly at pipelines but I think my use case is 'shorter' than that

I just a have a (potentially large) set of parallel calls

the results of which I want in a list

This must be a common pattern for core.async usage I just wasn't succeeding in finding any good examples

well, the moment you’re creating that final list, you’ll have to do it in 1 process right?

yes, there is a consumer of that list

but if you want to do processing on the values of the list, you might consider doing that before you create the list

and then pipelines come in handy

yes, I run some xpath queries over the results as they come in

I see what you mean

I think I was just struggling to visualise the 'structure' of that

there's the fanout of the queries, the processing of the results and collation of the derived data

typically you also want to control the parallelism in the api calls by issuing them from a pipeline-async and then send the results to a pipeline with a transducer that performs the xpath queries

and at the end create the final list with a timeout

a ha

looks like something along the right lines

There is the canonical core.async example somewhere, hang on

http://swannodette.github.io/2013/07/12/communicating-sequential-processes/

search for fake-search

(let [in (chan)
      responses (chan)
      query-results (chan)]
  (pipeline-async n1 responses perform-api-request in)
  (pipeline n2 query-results query-xf responses)
  (onto-chan in requests)
  (go-loop []
           ;; over query-results
           ))

thanks stijn & slipset for those examples