Thanks @alexmiller this is really helpful.

https://pastebin.com/zsJf4N9L how can i start next sync AFTER first one? How to know first one finished?

@kwladyka the first dosync doesn't close until it's body completes, but put! is async

oh, wait, you are using >!! - so it doesn't return until all the messages are put

yeah but i have to be sure all jobs are done before run next one

at that moment i am not sure, because something can still processing

(archai/fetch-epoch #(>!! out-elastic %) input) - this line is not one time >!! it is doing this many times inside this function

so at right moment i have to run:

(doseq [epoch epochs]
        (>!! in-archai {:url (archai/make-url {:stream "BAR"
                                               :epoch epoch})}))

if i will run it immediately after first one it can be situation like: first on still processing and >!! out-elastic and second one start doing this at the some time.

because there is more than 1 worker

@kwladyka what about making two pipeline-async calls, closing the input channel on the first after the first doseq, then waiting on its output (which means all buffered results have completed), then starting up another one to use in the second doseq

whatever solution, but how can i know first one finished?

by waiting on it's return value right?

it will close the to channel - so I guess you also need a second to-channel

and iirc it returns its to-channel (for convenience)

i tried something like that (<!! (pipeline-async workers-archai out-elastic archai->elastic in-archai)) but it hangs here

did you close its input channel when you finished the puts?

it doesn't close its output until the input is closed and all results from it are delivered

oh… so maybe after

(doseq [epoch epochs]
      (>!! in-archai {:url (archai/make-url {:stream "FOO"
                                             :epoch epoch})}))

Am i right?

like it finish after finish processing all buffered queue not immediately after close channel?

right - you probably need to reorder things somewhat

great, thx

and it's slightly more complicated because if you close in-archai that means it will close out-elastic, and the function that consumes from out-elastic is the one that will detect that

the more things we need to duplicate here, the more it looks like you need to make a function that gets called twice instead of doing things twice in one function

yes, probably now i should a little refactor it

BTW (dotimes [_ workers-elastic] do you think this method to create N workers is ok? Any better way?

it's how I'd do it

(<!! (pipeline-async workers-archai out-elastic archai->elastic in-archai)) hmm this one return immediately when i (close! in-archai), not wait to the end of out-elastic

oh, hmm...

in right situation i should also know when workers will finish processing out-elastic, not only when out-elastic is empty

async is sometimes so confuse

honestly I'm sick and not totally with it today and should stop trying to help people with code, but yeah, async is intrinsically hard

don’t stop help people 😛 😉

but as long as you are sick people should help you 🙂

I'm just realizing I'm not thinking clearly

heh i am sick today too, exactly i for some reason i have extremely allergy today, like i have never had

so i perfectly understand “not thinking clearly” today 🙂