(ns ird.p2p.async
  "The channel contract every ird.p2p function follows, plus the tools to
  work with it.

  Every asynchronous function returns a core.async promise-chan that
  delivers exactly one value: the result (a plain map, vector, string, ...)
  or an ExceptionInfo describing the failure (see ird.p2p.errors). Nothing
  is thrown out of an SDK function; errors travel as values so a `go` block
  can decide what to do with them:

    (go (let [r (<! (p2p/join app secret))]
          (if (error? r) (handle r) (use r))))

  Inside a go block `<?` takes and rethrows errors, so ordinary try/catch
  works; `go-try` is a go block that catches whatever its body throws and
  delivers it as a value. Long-lived streams (room events, transport
  frames) are ordinary channels fed through `unbounded-chan`, so a slow
  consumer never blocks the producer and nothing is dropped."
  (:require-macros [ird.p2p.async])
  (:require [cljs.core.async :as async :refer [go go-loop <! >!]]
            [ird.p2p.impl.node :as node]))

(defn error?
  "True when V is an error value (any JavaScript Error, which includes
  ExceptionInfo)."
  [v]
  (node/error? v))

(defn throw-err
  "Throw V when it is an error value, else return it. `<?` expands to this."
  [v]
  (if (error? v) (throw v) v))

(defn unbounded-chan
  "An unbounded queue in front of a channel. Returns
  {:out <channel> :put! (fn [v]) :close! (fn [])}: `put!` never blocks and
  never drops, `out` delivers the values in order, and `close!` closes
  `out` once everything queued has been taken."
  []
  (let [q (atom #queue [])
        wake (async/chan (async/sliding-buffer 1))
        out (async/chan)
        done (atom false)]
    (go-loop []
      (let [v (peek @q)]
        (cond
          (some? v) (do (swap! q pop)
                        (when (>! out v) (recur)))
          @done (async/close! out)
          :else (do (<! wake) (recur)))))
    {:out out
     :put! (fn [v] (swap! q conj v) (async/offer! wake true) nil)
     :close! (fn [] (reset! done true) (async/close! wake) nil)}))

(defn promise->chan
  "Adapt a JS Promise to the channel contract."
  [p]
  (node/promise->chan p))

(defn ->promise
  "Adapt an ird.p2p channel to a JS Promise: resolves with the delivered
  value, rejects when that value is an error — for the edge of a
  :node-library build."
  [ch]
  (node/chan->promise ch error?))

(defn sleep
  "The default :sleep-fn — a channel that closes after MS milliseconds."
  [ms]
  (async/timeout (max 0 ms)))

(defn deliver!
  "Put V on the promise-chan CH (closing it when V is nil)."
  [ch v]
  (if (nil? v) (async/close! ch) (async/put! ch v))
  nil)