@jgdavey don't know your use case for parallelizing work but I can highly recommend the Claypool library for maximum multithreaded flexibility and speed

Hello all

I have this code

even though I have 1000 pages, I'm getting only 8 results. What am I doing wrong?

@dimovich you only read from the channel once, shouldn't you be reading it repeatedly until it gets closed by the pipeline?

that is, s/reading/taking/g

@noisesmith I thought <!! will close when there is no more data on the pipeline

it reads a single result

how can I read all of them?

I tried async/reduce, but same thing

by repeatedly consuming in a loop or go-loop, or using a construct that fills a collection from a channel

probably the simplest thing is (async/into [] c)

trying them now, thanks

the names of the functions sort of lead me to believe they are doing (likely blocking) io, which means you should be using pipeline-blocking or pipeline-async

yeah, http requests

it's all there in the docstring for pipeline '''

If you have multiple
  blocking operations to put in flight, use pipeline-blocking instead,
  If you have multiple asynchronous operations to put in flight, use
  pipeline-async instead.

(async/<!! (async/into [] (async/pipeline-blocking 4 out-chan xform in-chan))) still gives me 8 results

what makes you think you should be getting more than 8?

I have (async/onto-chan in-chan pages). And there are 1000 items in pages.

(partial map) is just map

yeah, right 🙂

are you sure there are 1000 items in pages?

are your transducers throwing items away?

no, they either modify, or return original

what happens if you replace xform with identity?

oh

you are taking from the channel return from the call to pipeline-blocking, not the output channel you give to pipeline blocking

the channel that pipeline-blocking returns is nothing

@hiredman: I thought I'd checked in the past and it returned the output channel

I am pretty sure it doesn't

@hiredman yay, it works 🙂 thanks

the fact that you go any number of response from the channel returned from pipeline-blocking instead of just getting a closed suggests some kind of bug in your code somewhere

maybe it didn't return anything. I was getting log entries from the update-stat fns

@noisesmith @hiredman thank you both for helping!

I'm wondering which core.async function I was thinking of, that allowed forwarding data from one chan to another with a transducer, which conveniently returns the destination channel

not pipeline* - none of them actually work that way

you can do (async/chan 1 xform)

right, but sometimes you have a channel that comes from someone else, and you want an xform on it