4cf95dc27c
Adds clojure.core.async's higher-level dataflow API as a Clojure overlay (stdlib/clojure/core/async.clj) over jolt's native channel primitives, plus clojure.core.async.lab. The native layer (host/chez/java/async.ss) gains offer!/poll!, put specs and :priority/:default in alts!, a transducer ex-handler arg to chan, unblocking-buffer?, promise-buffer, and on-caller? handling for put!/take!. The overlay covers alts!/pipe/pipeline/split/ reduce/transduce/into/take/mult/mix/pub-sub/map/merge/onto-chan/to-chan and the deprecated map</map>/filter>/... family (rewritten as go-loops since the JVM versions reify the impl handler protocol jolt doesn't expose). Loading: the native primitives pre-seed clojure.core.async, so the loader now drops it from the loaded set and a require pulls the overlay from the source roots like clojure.test (AOT-bundled into built binaries). Running clojure/core.async's own suite shook out two general bugs: - :refer with a list form, (:require [ns :refer (a b c)]), dropped the names (only the vector form was handled) — chez-register-spec! now accepts both. - (range 0) / (range 5 5) returned nil instead of the empty seq () — empty ranges now match Clojure, so (= () (range 0)) holds. Suite: async_test 15/20, pipeline_test 7/7, timers_test 2/2, lab_test 2/2. The five non-passing async_test cases all assert JVM go-machine limitations jolt's thread-based model is a superset of (the 1024 pending-op cap, parking ops that must throw outside a go block, expanding-transducer buffer backpressure) or dispatch-thread identity, not data semantics. make test green (0 new divergences, +4 range corpus rows), shakesmoke byte-identical.
34 lines
1.1 KiB
Clojure
34 lines
1.1 KiB
Clojure
;; clojure.core.async.lab — experimental features over the channel primitives.
|
|
;;
|
|
;; multiplex/broadcast are ported as go-loops over jolt's primitives (the JVM
|
|
;; versions reify the impl handler protocol, which jolt does not expose).
|
|
|
|
(ns clojure.core.async.lab
|
|
(:require [clojure.core.async :as async]))
|
|
|
|
(defn multiplex
|
|
"Returns a read port that yields values from whichever of ports is ready. A
|
|
closed port is dropped; the multiplex port closes once all ports have closed."
|
|
[& ports]
|
|
(let [out (async/chan)]
|
|
(async/go-loop [cs (vec ports)]
|
|
(if (pos? (count cs))
|
|
(let [[v c] (async/alts! cs)]
|
|
(if (nil? v)
|
|
(recur (filterv #(not= c %) cs))
|
|
(do (async/>! out v)
|
|
(recur cs))))
|
|
(async/close! out)))
|
|
out))
|
|
|
|
(defn broadcast
|
|
"Returns a write port that writes each value to all of ports. A write parks until
|
|
the value has been written to every port."
|
|
[& ports]
|
|
(let [in (async/chan)]
|
|
(async/go-loop []
|
|
(let [v (async/<! in)]
|
|
(when (some? v)
|
|
(doseq [p ports] (async/>! p v))
|
|
(recur))))
|
|
in))
|