how is it different from any other task ?

I get back an Event from the rdv, and I want to get its result

oh I misread I thought it was a mbx

is there a reason why you need rdv here ?

That's based on your suggestion from when I tried to convert callback based functions to ask, like promise or completable future

I don't remember exactly what I suggested but this is clearly a bad usage of rdv

rdv with 1 argument is pure and returns a task that puts the value and completes when it's read

https://github.com/leonoel/missionary/issues/45

here I suggest dfv

Same semantics?

Ah, no delay

Using [java-http-clj.core :as http]

I'm working on building an end to end example with missionary in a web programming context

I'd love to see that

For http I think in practice it's OK to simply wrap the callback API and ignore cancellation, assuming the timeouts are reasonably low

(def flow-dad
  (fn [s f]
    (http/send-async
      {:uri     ""
       :headers {"accept" "text/plain"}
       :method  :get}
      nil s f) #()))

I'm assuming a context where errors have to be handled, but I'll try this as well to figure out what I missed. For some reason this didn't work for me, but maybe something small was off

oh, was it because I was returning something I shouldn't have been?

I needed to return a cancelling function?

yes, it's mandated by the task protocol

That's what I forgot Good, now I'm at:

(defn prn-rf
    ([])
    ([_])
    ([_ x] (println x)))

  (m/?
   (m/reduce
    prn-rf
    (m/eduction
     (comp (take 2) (map :body))
     (forever dad-joke-task))))


  (->>
   (m/seed ["R7UfaahVfFd" "q4hVKRnOmjb" "xHQucUvszd"])
   (m/eduction (map (comp m/? joke-task)))
   (m/eduction (map :body))
   (m/reduce prn-rf)
   m/?)

Both work nicely

next step is fork/joining over a seed of tasks and collecting the results

(m/eduction (map (comp m/? joke-task))) m/? is blocking the thread here, is it your intent ?

no

I didn't know how to "flatmap" the tasks

I would put everything in an ap

Everything = eduction or even the reduce?

(m/ap
  (-> (m/?> (m/seed ["R7UfaahVfFd" "q4hVKRnOmjb" "xHQucUvszd"]))
    joke-task m/? :body println))

ack, I forget that inside the ap the context changes to a single value from flow

println in the reducing function is OK and maybe more modular

Here we go!

(->> ["R7UfaahVfFd" "q4hVKRnOmjb" "xHQucUvszd"]
      m/seed
      m/?>
      joke-task
      m/?
      :body
      m/ap
      (m/reduce conj)
      m/?)

😎

(defn map-reduce
    [rf f flow]
    (->> flow m/?> f m/? m/ap (m/reduce rf)))

  (->> ["R7UfaahVfFd" "q4hVKRnOmjb" "xHQucUvszd"]
       m/seed
       (map-reduce ((map :body) conj) joke-task)
       m/?)

To the cookbook?

maybe ap in the threading macro is a bit too much for beginners

but this is the right way

Yeah it needs to be organized such that the boundaries of macro magic are clearer

Slightly more readable

(defn map-reduce
    [rf f flow]
    (->>
     (m/ap (->> flow m/?> f m/?))
     (m/reduce rf)))

I like this one

Is this only correct for f which takes data and returns a task?

yes

So this is mapcat ?

(m/ap (->> flow m/?> f m/?))

not exactly, because there are 2 type constructors here

Can you please elaborate? Also, is it more correct to say this is monadic bind?

the signature is (A -> Task[B]) -> Flow[A] -> Flow[B]

monadic bind would be (A -> Flow[B]) -> Flow[A] -> Flow[B]

it is called concatMapSingle in RxJava http://reactivex.io/RxJava/3.x/javadoc/io/reactivex/rxjava3/core/Flowable.html#concatMapSingle-io.reactivex.rxjava3.functions.Function-

It's just an aesthetic preference but I dislike the names in Rx. Is the transformation of A -> T[B] to A -> F[B] a Lift?

that's not how I understand lift, but I may be wrong

more likely I am wrong, my cat theory is weak

Looking in wiki (>>=) : (M T, T → M U) → M U so if `mx : M T` and `f : T → M U`, then `(mx >>= f) : M U` Then the transition from T[B] to F[B] is a bind

how it works or what it's for ?

probably both

the tests are also very sparse and tell me very little

do you understand the docstring ?

i think so

basically it works like take-nth ?

((mi/reduce conj (mi/ap
                   (let [x (mi/?< (mi/buffer 3 (mi/seed (range 12))))]
                     x))) prn prn)
;=> [0 3 6 9]

is identical to

((mi/reduce conj (mi/ap
                   (let [x (mi/?< (mi/eduction (take-nth 3) (mi/seed (range 12))))]
                     x))) prn prn)

no

do you understand what backpressure is about ?

yes

good

buffer is a pipeline stage that doesn't change the semantics of the flow, it just consumes more memory.

I've tried attaching a buffer to relieve somehow, or to faster input than output, but I don't have any conclusions as to what this does

It is probably possible to do dropping buffer and sliding buffer like in core.async, but how?

that is a different concern

buffer doesn't solve that, it is just allocating more storage to absorb temporary fast inputs

it is the same as core.async channel capacity

:thinking_face: