Some fun with callbacks

callbacks-are-fn.clj

;; Given a seq of something and an async fn that accepts a callback,
;; how would you process the seq synchronously in order?

(def atm (atom []))
(reset! atm [])
(add-watch atm
           :watcha
           (fn [k r o n]
             (prn k o n)))

;; Some async action that accepts a callback
(defn async-thing [callback]
  (future (Thread/sleep 1000)
          (callback)))

;; 'async', order of elements will likely be out of order
(doseq [el (range 10)]
  (async-thing (fn [] (swap! atm #(conj % el)))))

;; Looks synchronous, but elements get processed in reverse order
((loop [[head nxt & left] (range 10)
        head-clbk (fn []
                    (async-thing (fn [] (swap! atm #(conj % head)))))]
   (if nxt
     (recur `[~nxt ~@left] (fn [] (async-thing
                                   (fn []
                                     (head-clbk)
                                     (swap! atm #(conj % nxt))))))
     head-clbk)))

;; Switching nxt/head does not help!
;; Is there a way to process elements in order without reversing the seq?
;; Is core.async a tool that could be useful in this situation?
((loop [[head nxt & left] (range 10)
        head-clbk (fn []
                    (async-thing (fn [] (swap! atm #(conj % nxt)))))]
   (if nxt
     (recur `[~nxt ~@left] (fn [] (async-thing
                                   (fn []
                                     ;; Note switching order of these two does not
                                     ;; affect the result as (head-clbk) returns right away.
                                     (head-clbk)
                                     (swap! atm #(conj % head))))))
     head-clbk)))


;; Using a channel with watches,
;; we can achieve one element at atime & preserve order
(def chan (atom #{}))
(reset! atm [])

(loop [[el next-el & left] (range 100)]
  (add-watch chan
             el ;;The element to watch for
             (fn [k r o n]
               ;; Only want to fire the task when the
               ;; watching key has been newly conj'ed
               (when (and (get n k) (not (get o k)))
                 (async-thing
                  (fn []
                    (prn "doing async task for..." k)
                    (swap! atm #(conj % el))

                    ;; Triggers the watch for next-el
                    (swap! chan #(conj % next-el)))))))
  (if next-el
    (recur `[~next-el ~@left])
    nil))

;; Add the first element, and the watches start to cascade
(swap! chan #(conj % 0))

You can block callback-style async processes by using promise:

(time (->> (range 10)
      (map (fn [el]
             (let [p (promise)]
               (async-thing (fn []
                              (let [ret (* el 2)]
                                (println ret)
                                (deliver p ret))))
               @p)))
      doall))
;; 0
;; 2
;; 4
;; 6
;; 8
;; 10
;; 12
;; 14
;; 16
;; 18
;; "Elapsed time: 10021.609412 msecs"
;; => (0 2 4 6 8 10 12 14 16 18)

Even if you are concerned only about the order of return values, it also can be achieved with promise:

(time (->> (range 10)
           (map (fn [el]
                  (let [p (promise)]
                    (async-thing (fn []
                                   (let [ret (* el 2)]
                                     (deliver p ret))))
                    p)))
           (map deref)
           doall))
;; "Elapsed time: 1013.232371 msecs"
;; => (0 2 4 6 8 10 12 14 16 18)

I might have misunderstood your point. If you want to keep the callback style and avoid blocking, you can use monadic approach:

(defn cont-pure [v]
  (fn [cb]
    (cb v)))

(defn cont-flatmap [mv f]
  (fn [cb]
    (mv (fn [v]
          ((f v)
           (fn [v']
             (cb v')))))))

(defn async-doubler [i cb]
  (future
    (Thread/sleep (rand-int 1000))
    (cb (* i 2)))
  nil)

(def chained-async-doubler (->> (range 10)
                                (reduce (fn [acc _]
                                          (cont-flatmap acc
                                                        (fn [v]
                                                          (println v)
                                                          (fn [cb]
                                                            (async-doubler v cb)))))
                                        (cont-pure 1))))

(chained-async-doubler #(println "done" %))
;; => nil
;; Output:
;; 1
;; 2
;; 4
;; 8
;; 16
;; 32
;; 64
;; 128
;; 256
;; 512
;; done 1024

@rfkm Thanks! Promise is what I was looking for. Was not aware of that :)

And keeping order + async was something that I was curious, so thanks for that too!