Cancelable Asyn

async.clj

(defprotocol Async
  (run-async* [this continue])
  (cancel-async* [this]))

(defn await-event
  "Awaits the given event on the given target asynchronuously. Passes on the
  event to the continuation."
  [target event]
  (let [remove-fn (promise)]
    (reify Async
      (run-async* [this continue]
        (let [handler   (fn [evt] (@remove-fn) (continue evt))]
          (deliver remove-fn (listen target event handler))))
      (cancel-async* [this] (@remove-fn))
      
(defn wait
  "Wait asynchronuously for t milliseconds. Passes on nil to the continuation."
  [millis]
  (let [t (atom nil)]
    (reify Async
      (run-async* [this continue]
        (reset! t (timer (fn [_] (continue nil))
                    :initial-delay millis
                    :repeats?      false)))
      (cancel-async* [this] (if @t (.stop @t))))))

(defn await-future*
  "Call the function with any additional arguments in a background thread and
  wait asynchronuously for its completion. Passes on the result to the
  continuation.
  See also: await-future"
  [f & args]
  (let [canceled? (atom false)]
    (reify Async
      (run-async* [this continue]
        (future
          (let [result (apply f args)]
            (if-not @canceled?
              (invoke-now
                (continue result))))))
      (cancel-async* [this] (reset! canceled? true)))))

(defmacro await-future
  "Execute the code block in a background thread and wait asynchronuously
  for its completion. Passes on the result to the continuation.
  See also: await-future*"
  [& body]
  `(await-future* (fn [] ~@body)))

(defn await-any
  "Wait asynchronuously until one of the given events happens. Passes on
  the result of the triggering event to the continuation."
  [& events]
  (let [canceled? (atom false)]
    (reify Async
      (run-async* [this global-continue]
        (let [guard          (atom false)
              inner-continue (fn [result]
                              (when (compare-and-set! guard false true)
                                (global-continue result)))]
          (if-not @canceled?
            (doseq [event events] (run-async* event inner-continue)))))
      (cancel-async* [this] 
        (reset! canceled? true)
        (doseq [event events] (cancel-async* event))))))