You are trying to use pipeline-async with clj-http. To do that, use async HTTP calls and supply a callback method that will async/put! the result on the out channel provided by pipeline-async

thanks, looking at that too. @U0NCTKEV8 also had some pointers about redis reading.

so I have a working version, thanks for all the pointers:

I am not sure I understand why I need to put! to out and close out* . I believe closing out* is used for signaling. cose = work has finished for the url ?!

Well first because this is the contract that pipeline-async specifies for the af async function: >

af
> must be a function of two arguments, the first an input value and
> the second a channel on which to place the result(s). af must close!
> the channel before returning.

I updated my solution to work properly regarding out channel . I was not putting to the channel received from pipeline async. thanks

You should be writing to out* then closing out*

yes, I fixed that just now

I think I am getting the hang of this - could not have done it without your help. I will try to migrate the sample to use pipeline-blocking.

In the exception case you are not closing out*

good point

(defn request-handler [url out*]
    (async/go

Also, something you should never forget to do in core.async (I learned it the hard way): You need to ensure that your application will wait for async processes to finish. In your case, the pipeline-blocking will return a channel that will close when done. If you call this from -main without blocking the main thread on this channel, your application will immediately end without waiting for the async processes to finish.

I fixed the function.

for pipeline-blocking I have this code. But it put! a sequence to the channel instead of individual values. I guess the semantics are changed in this case and I will have to update my code to take a sequence from the out channel. Just asking: is there a way to put! individual values in the case of pipeline-blocking ?!

(async/go
  (async/>! out* (get repo "clone_url"))

In the pipeline-blocking case: Nope, it will put the return value on the out channel. But you can use a transducer with your out channel:

(let [out (async/chan 4 cat)]

io mean in the handler function I should have a go loop instead of a go?

I don't understand what you mean, could you please rephrase it ?

You are doing blocking io reading from redis in a go, which is blocking the threadpool that go's run on

You are not supposed to do blocking operations in go because it will starve other go blocks

I don't think that is the case. This code prints "Sleeping" every 1s since car/lpop returns nil if there is no value

(async/go-loop []
      (if-let [message (wcar* (car/lpop "queue"))]
        (do
          (println "Got message from queue" message)
          (async/>! in message))
      ; Sleep if no messages available
        (do
          (println "sleeping")
          (async/<! (async/timeout 1000))))
      (recur))

If it is the cause of your issue or not I don't know, but it is absolutely the case that you should not do it, and it will cause issues if you do

thanks, I will take a look at this as well - seems redis is synchronous by nature https://github.com/ptaoussanis/carmine/issues/148 .

It will depend on the redis client you use, but in general there is no problem doing blocking io, just don't do it directly on a go

I use carmine

Wrap your io in the thread macro

Carmine internally runs a global threadpool, which is not great

thanks, I did not know that. I'm exploring now so that is not important atm. This is what I have for now

(async/thread
      (loop
       (if-let [message (wcar* (car/lpop "queue"))]
         (do
           (println "Got message from queue" message)
           (async/>!! in message))
          ; Sleep if no messages available
         (Thread/sleep 1000))
        (recur)))

Redis has a whole pubsub thing