Merge pull request #253 from jolt-lang/conformance/core-async
core.async: higher-level API over native channels + general fixes
This commit is contained in:
Dmitri Sotnikov
GitHub
commit
c479010536
8 changed files with 834 additions and 57 deletions
|
|
@ -48,6 +48,10 @@ e.g. the [ring-app example](https://github.com/jolt-lang/examples/tree/main/ring
|
|||
[data.priority-map](https://github.com/clojure/data.priority-map).
|
||||
* [core.memoize](https://github.com/clojure/core.memoize) — function memoization
|
||||
over [core.cache](https://github.com/clojure/core.cache).
|
||||
* [core.async](https://github.com/clojure/core.async) — CSP channels and `go` blocks
|
||||
(`<!`/`>!`/`alts!`, `pipeline`, `mult`/`mix`/`pub`/`sub`) on real OS threads.
|
||||
* [core.logic](https://github.com/clojure/core.logic) — relational logic programming
|
||||
(unification, `run`/`fresh`/`conde`, finite domains).
|
||||
* [tick](https://github.com/juxt/tick) — date/time over Jolt's `java.time`;
|
||||
`#time/…` literals via `time-literals`.
|
||||
* [transit-jolt](https://github.com/jolt-lang/transit-jolt) — Transit (JSON) read/write
|
||||
|
|
|
|||
|
|
@ -1,17 +1,20 @@
|
|||
;; async.ss — clojure.core.async on real OS threads for the Chez host.
|
||||
;; async.ss — clojure.core.async channel primitives on real OS threads.
|
||||
;;
|
||||
;; A `go` block is an OS thread and a channel is a mutex+condition blocking
|
||||
;; queue: <! / >! are the blocking <!! / >!! (they "park" by blocking the thread).
|
||||
;; <! / >! work ANYWHERE — no CPS transform — because they are ordinary blocking
|
||||
;; calls. Real parallelism, shared heap. Trade-off: one OS thread per go block
|
||||
;; (fine for typical use, not for thousands of simultaneous go blocks).
|
||||
;; A `go` block is an OS thread and a channel is a Chez mutex+condition blocking
|
||||
;; queue: <! / >! are the blocking <!! / >!! (they "park" by blocking the thread),
|
||||
;; and work ANYWHERE — no CPS transform, no go-only restriction. Real parallelism,
|
||||
;; shared heap. This is a superset of the JVM model: it has no fixed go-block
|
||||
;; thread pool, no MAX-QUEUE-SIZE on pending ops, and parking ops are legal outside
|
||||
;; a go block. One OS thread per go block (fine for typical use).
|
||||
;;
|
||||
;; Channel: an unbuffered channel is a rendezvous (the putter blocks until its
|
||||
;; value is taken); a buffered (chan n) put blocks only when full; dropping/sliding
|
||||
;; buffers never block the putter. A transducer is applied on the put side.
|
||||
;; buffers never block the putter. A transducer is applied on the put side; an
|
||||
;; optional ex-handler catches a throw from the transducer step.
|
||||
;;
|
||||
;; The fns are def-var!'d into clojure.core.async; go/go-loop/thread are macros
|
||||
;; (mark-macro!) expanding to go-spawn. Loaded after
|
||||
;; This file provides the primitives; the higher-level dataflow API (mult, mix,
|
||||
;; pub/sub, pipeline, map, merge, reduce, …) is a Clojure overlay over them.
|
||||
;; go/go-loop/thread are macros (mark-macro!) expanding to go-spawn. Loaded after
|
||||
;; concurrency.ss (reuses ms->duration). Requires a threaded Chez build.
|
||||
|
||||
;; --- buffers ----------------------------------------------------------------
|
||||
|
|
@ -19,6 +22,8 @@
|
|||
(define (jolt-async-buffer n) (make-async-buffer n 'fixed))
|
||||
(define (jolt-async-dropping-buffer n) (make-async-buffer n 'dropping))
|
||||
(define (jolt-async-sliding-buffer n) (make-async-buffer n 'sliding))
|
||||
(define (jolt-async-unblocking-buffer? b)
|
||||
(if (and (async-buffer? b) (memq (async-buffer-kind b) '(dropping sliding promise))) #t #f))
|
||||
|
||||
;; --- channels ---------------------------------------------------------------
|
||||
;; items: an amortized-O(1) FIFO held as a mutable #(out in len) — `out` is the
|
||||
|
|
@ -27,9 +32,12 @@
|
|||
;; Each entry is (value . box); box is #f for a buffered value or a 1-slot vector
|
||||
;; for an unbuffered rendezvous put (set #t when taken, waking the putter).
|
||||
;; cap 0 + kind 'unbuffered = rendezvous; cap>0 with kind fixed/dropping/sliding.
|
||||
;; takew counts threads parked in a blocking take (so a non-blocking offer! to an
|
||||
;; unbuffered channel can tell a taker is waiting). xrf is the transducer reducing
|
||||
;; fn (or #f); exh the ex-handler (or #f).
|
||||
(define-record-type async-chan
|
||||
(fields mu cv (mutable items) cap kind (mutable closed?) (mutable xrf))
|
||||
(nongenerative async-chan-v1))
|
||||
(fields mu cv (mutable items) cap kind (mutable closed?) (mutable xrf) (mutable takew) exh)
|
||||
(nongenerative async-chan-v2))
|
||||
|
||||
(define (ac-qnew) (vector '() '() 0))
|
||||
(define (ac-qlen ch) (vector-ref (async-chan-items ch) 2))
|
||||
|
|
@ -73,17 +81,30 @@
|
|||
((null? (cdr args)) (car args)) ; completion
|
||||
(else (ac-buf-give! ch (cadr args)) (car args))))) ; step
|
||||
|
||||
(define (ac-make cap kind xrf) (make-async-chan (make-mutex) (make-condition) (ac-qnew) cap kind #f xrf))
|
||||
;; run the transducer step (or completion) guarded by the channel's ex-handler:
|
||||
;; if the xform throws and exh returns non-nil, that value is added to the buffer.
|
||||
(define (ac-xrf-apply ch . v)
|
||||
(let ((xrf (async-chan-xrf ch)) (exh (async-chan-exh ch)))
|
||||
(guard (e (#t (if exh
|
||||
(let ((else (jolt-invoke exh e)))
|
||||
(unless (jolt-nil? else) (ac-buf-give! ch else))
|
||||
(async-chan-xrf ch)) ; treat as non-reduced
|
||||
(raise e))))
|
||||
(apply jolt-invoke xrf ch v))))
|
||||
|
||||
;; (chan) | (chan n) | (chan buf) | (chan n|buf xform)
|
||||
(define (ac-make cap kind xrf) (make-async-chan (make-mutex) (make-condition) (ac-qnew) cap kind #f xrf 0 #f))
|
||||
(define (ac-make/exh cap kind exh) (make-async-chan (make-mutex) (make-condition) (ac-qnew) cap kind #f #f 0 exh))
|
||||
|
||||
;; (chan) | (chan n) | (chan buf) | (chan n|buf xform) | (chan n|buf xform exh)
|
||||
(define (jolt-async-chan . args)
|
||||
(let ((buf (if (pair? args) (car args) jolt-nil))
|
||||
(xform (if (and (pair? args) (pair? (cdr args))) (cadr args) jolt-nil)))
|
||||
(xform (if (and (pair? args) (pair? (cdr args))) (cadr args) jolt-nil))
|
||||
(exh (if (and (pair? args) (pair? (cdr args)) (pair? (cddr args))) (caddr args) jolt-nil)))
|
||||
(let-values (((cap kind)
|
||||
(cond ((async-buffer? buf) (values (async-buffer-n buf) (async-buffer-kind buf)))
|
||||
((and (number? buf) (> buf 0)) (values buf 'fixed))
|
||||
(else (values 0 'unbuffered)))))
|
||||
(let ((ch (ac-make cap kind #f)))
|
||||
(let ((ch (ac-make/exh cap kind (if (jolt-nil? exh) #f exh))))
|
||||
(unless (jolt-nil? xform)
|
||||
(async-chan-xrf-set! ch (jolt-invoke xform (ac-make-add-rf ch))))
|
||||
ch))))
|
||||
|
|
@ -93,7 +114,7 @@
|
|||
(define (ac-close! ch)
|
||||
(unless (async-chan-closed? ch)
|
||||
(async-chan-closed?-set! ch #t)
|
||||
(when (async-chan-xrf ch) (guard (e (#t #f)) (jolt-invoke (async-chan-xrf ch) ch)))
|
||||
(when (async-chan-xrf ch) (guard (e (#t #f)) (ac-xrf-apply ch)))
|
||||
(condition-broadcast (async-chan-cv ch)))
|
||||
jolt-nil)
|
||||
(define (jolt-async-close! ch) (with-mutex (async-chan-mu ch) (ac-close! ch)))
|
||||
|
|
@ -102,12 +123,12 @@
|
|||
;; transducer the value is run through it (one put -> zero or more channel values);
|
||||
;; a `reduced` result closes the channel.
|
||||
(define (jolt-async-give ch v)
|
||||
(when (jolt-nil? v) (jolt-throw (jolt-ex-info "Can't put nil on a channel" (jolt-hash-map))))
|
||||
(when (jolt-nil? v) (jolt-throw (jolt-host-throwable "java.lang.IllegalArgumentException" "Can't put nil on a channel")))
|
||||
(with-mutex (async-chan-mu ch)
|
||||
(cond
|
||||
((async-chan-closed? ch) #f)
|
||||
((async-chan-xrf ch)
|
||||
(let ((r (jolt-invoke (async-chan-xrf ch) ch v)))
|
||||
(let ((r (ac-xrf-apply ch v)))
|
||||
(when (jolt-reduced? r) (ac-close! ch))
|
||||
#t))
|
||||
(else
|
||||
|
|
@ -154,12 +175,19 @@
|
|||
(cond ((eq? (async-chan-kind ch) 'promise)
|
||||
(cond ((not (ac-qempty? ch)) (ac-peek ch))
|
||||
((async-chan-closed? ch) jolt-nil)
|
||||
(else (condition-wait (async-chan-cv ch) (async-chan-mu ch)) (loop))))
|
||||
(else (ac-take-wait ch) (loop))))
|
||||
((not (ac-qempty? ch)) (ac-take-head! ch))
|
||||
((async-chan-closed? ch) jolt-nil)
|
||||
(else (condition-wait (async-chan-cv ch) (async-chan-mu ch)) (loop))))))
|
||||
(else (ac-take-wait ch) (loop))))))
|
||||
|
||||
;; non-blocking take for alts!: a value, jolt-nil (closed+empty), or ac-poll-empty.
|
||||
;; park in a take, tracking the waiter count so a concurrent offer! to an
|
||||
;; unbuffered channel can see that a taker is ready.
|
||||
(define (ac-take-wait ch)
|
||||
(async-chan-takew-set! ch (fx+ 1 (async-chan-takew ch)))
|
||||
(condition-wait (async-chan-cv ch) (async-chan-mu ch))
|
||||
(async-chan-takew-set! ch (fx- (async-chan-takew ch) 1)))
|
||||
|
||||
;; non-blocking take for alts!/poll!: a value, jolt-nil (closed+empty), or ac-poll-empty.
|
||||
(define ac-poll-empty (list 'empty))
|
||||
(define (ac-poll! ch)
|
||||
(with-mutex (async-chan-mu ch)
|
||||
|
|
@ -168,28 +196,40 @@
|
|||
((async-chan-closed? ch) jolt-nil)
|
||||
(else ac-poll-empty))))
|
||||
|
||||
;; (alts! [ch ...]) — take from whichever channel is ready first; returns
|
||||
;; [value channel] (value nil if that channel closed). Take-only: every port must
|
||||
;; be a channel — put specs [ch val] and the :default option are not supported, so
|
||||
;; reject them with a clear error instead of crashing inside ac-poll!.
|
||||
;; Polls with a 1ms backoff — no cross-channel wait-set yet.
|
||||
(define ac-1ms (make-time 'time-duration 1000000 0))
|
||||
(define (jolt-async-alts chans)
|
||||
(let ((cs (seq->list (jolt-seq chans))))
|
||||
(for-each (lambda (c)
|
||||
(unless (async-chan? c)
|
||||
(jolt-throw (jolt-ex-info
|
||||
"alts! supports channel ports only (put specs [ch val] and :default are not supported)"
|
||||
(jolt-hash-map)))))
|
||||
cs)
|
||||
(let loop ()
|
||||
(let try ((rest cs))
|
||||
(if (null? rest)
|
||||
(begin (sleep ac-1ms) (loop))
|
||||
(let ((r (ac-poll! (car rest))))
|
||||
(if (eq? r ac-poll-empty)
|
||||
(try (cdr rest))
|
||||
(jolt-vector r (car rest)))))))))
|
||||
;; non-blocking give: 'ok (accepted), 'full (would block), or 'closed.
|
||||
(define (ac-try-give! ch v)
|
||||
(when (jolt-nil? v) (jolt-throw (jolt-host-throwable "java.lang.IllegalArgumentException" "Can't put nil on a channel")))
|
||||
(with-mutex (async-chan-mu ch)
|
||||
(cond
|
||||
((async-chan-closed? ch) 'closed)
|
||||
((async-chan-xrf ch) (let ((r (ac-xrf-apply ch v)))
|
||||
(when (jolt-reduced? r) (ac-close! ch)) 'ok))
|
||||
(else
|
||||
(case (async-chan-kind ch)
|
||||
((dropping sliding) (ac-buf-give! ch v) 'ok)
|
||||
((promise) (when (ac-qempty? ch) (ac-qpush! ch (cons v #f))
|
||||
(condition-broadcast (async-chan-cv ch))) 'ok)
|
||||
(else
|
||||
(cond
|
||||
((> (async-chan-cap ch) 0)
|
||||
(if (< (ac-qlen ch) (async-chan-cap ch))
|
||||
(begin (ac-qpush! ch (cons v #f)) (condition-broadcast (async-chan-cv ch)) 'ok)
|
||||
'full))
|
||||
;; unbuffered: only immediate if a taker is parked to receive it.
|
||||
((> (async-chan-takew ch) 0)
|
||||
(let ((box (vector #f)))
|
||||
(ac-qpush! ch (cons v box))
|
||||
(condition-broadcast (async-chan-cv ch))
|
||||
'ok))
|
||||
(else 'full))))))))
|
||||
|
||||
;; offer! / poll! — never block. offer! returns #t/#f(closed) on completion, nil if
|
||||
;; it would block; poll! returns a value, nil (closed+empty), or the ::none sentinel.
|
||||
(define cca-none (keyword "clojure.core.async" "none"))
|
||||
(define (jolt-async-offer! ch v)
|
||||
(case (ac-try-give! ch v) ((ok) #t) ((closed) #f) (else jolt-nil)))
|
||||
(define (jolt-async-poll! ch)
|
||||
(let ((r (ac-poll! ch))) (if (eq? r ac-poll-empty) cca-none r)))
|
||||
|
||||
;; (timeout ms) — a channel that closes after ms milliseconds.
|
||||
(define (jolt-async-timeout ms)
|
||||
|
|
@ -197,17 +237,28 @@
|
|||
(fork-thread (lambda () (sleep (ms->duration ms)) (jolt-async-close! w)))
|
||||
w))
|
||||
|
||||
;; (put! ch v [cb]) / (take! ch cb) — async put/take on a thread, optional callback.
|
||||
(define (jolt-async-put! ch v . cb)
|
||||
(fork-thread (lambda ()
|
||||
(let ((ok (jolt-async-give ch v)))
|
||||
(when (and (pair? cb) (not (jolt-nil? (car cb)))) (jolt-invoke (car cb) ok)))))
|
||||
jolt-nil)
|
||||
(define (jolt-async-take! ch cb)
|
||||
(fork-thread (lambda ()
|
||||
(let ((v (jolt-async-take ch)))
|
||||
(unless (jolt-nil? cb) (jolt-invoke cb v)))))
|
||||
jolt-nil)
|
||||
;; (put! ch v [cb [on-caller?]]) — async put, optional completion callback. If the
|
||||
;; put completes immediately and on-caller? (default #t), the callback runs on the
|
||||
;; calling thread; otherwise on another thread. Returns true unless already closed.
|
||||
(define (jolt-async-put! ch v . rest)
|
||||
(let* ((cb (if (pair? rest) (car rest) jolt-nil))
|
||||
(on-caller? (if (and (pair? rest) (pair? (cdr rest))) (jolt-truthy? (cadr rest)) #t))
|
||||
(call-cb (lambda (ok) (unless (jolt-nil? cb) (jolt-invoke cb ok)))))
|
||||
(case (ac-try-give! ch v)
|
||||
((ok) (if on-caller? (call-cb #t) (fork-thread (lambda () (call-cb #t)))) #t)
|
||||
((closed) (if on-caller? (call-cb #f) (fork-thread (lambda () (call-cb #f)))) #f)
|
||||
(else (fork-thread (lambda () (call-cb (jolt-async-give ch v)))) #t))))
|
||||
|
||||
;; (take! ch cb [on-caller?]) — async take. Same on-caller? rule as put!.
|
||||
(define (jolt-async-take! ch cb . rest)
|
||||
(let* ((on-caller? (if (pair? rest) (jolt-truthy? (car rest)) #t))
|
||||
(call-cb (lambda (v) (unless (jolt-nil? cb) (jolt-invoke cb v))))
|
||||
(r (ac-poll! ch)))
|
||||
(cond
|
||||
((eq? r ac-poll-empty) (fork-thread (lambda () (call-cb (jolt-async-take ch)))))
|
||||
(on-caller? (call-cb r))
|
||||
(else (fork-thread (lambda () (call-cb r)))))
|
||||
jolt-nil))
|
||||
|
||||
;; (go-spawn thunk) — run thunk on a thread; return a buffered(1) channel that
|
||||
;; conveys its value once then closes (a nil result just closes). Dynamic bindings
|
||||
|
|
@ -246,14 +297,19 @@
|
|||
(cca-def! "buffer" jolt-async-buffer)
|
||||
(cca-def! "dropping-buffer" jolt-async-dropping-buffer)
|
||||
(cca-def! "sliding-buffer" jolt-async-sliding-buffer)
|
||||
(cca-def! "__promise-buffer" (lambda () (make-async-buffer 1 'promise)))
|
||||
(cca-def! "unblocking-buffer?" jolt-async-unblocking-buffer?)
|
||||
(cca-def! "close!" jolt-async-close!)
|
||||
(cca-def! "<!" jolt-async-take) (cca-def! "<!!" jolt-async-take)
|
||||
(cca-def! ">!" jolt-async-give) (cca-def! ">!!" jolt-async-give)
|
||||
(cca-def! "alts!" jolt-async-alts) (cca-def! "alts!!" jolt-async-alts)
|
||||
(cca-def! "timeout" jolt-async-timeout)
|
||||
(cca-def! "put!" jolt-async-put!)
|
||||
(cca-def! "take!" jolt-async-take!)
|
||||
(cca-def! "offer!" jolt-async-offer!)
|
||||
(cca-def! "go-spawn" async-go-spawn)
|
||||
;; non-blocking primitives the Clojure overlay's do-alts polls over.
|
||||
(cca-def! "__poll!" jolt-async-poll!)
|
||||
(cca-def! "__offer!" jolt-async-offer!)
|
||||
(cca-def! "go" cca-go-macro) (mark-macro! "clojure.core.async" "go")
|
||||
(cca-def! "go-loop" cca-go-loop-macro) (mark-macro! "clojure.core.async" "go-loop")
|
||||
(cca-def! "thread" cca-thread-macro) (mark-macro! "clojure.core.async" "thread")
|
||||
|
|
|
|||
|
|
@ -141,6 +141,14 @@
|
|||
(vector-for-each (lambda (c) (hashtable-set! loaded-ns (var-cell-ns c) #t))
|
||||
(hashtable-values var-table))
|
||||
|
||||
;; clojure.core.async ships native channel primitives (async.ss) AND a Clojure
|
||||
;; overlay (stdlib/clojure/core/async.clj) with the higher-level dataflow API
|
||||
;; (alts!, pipe, mult, mix, pub/sub, map, merge, …). The primitives pre-seed the
|
||||
;; namespace above, which would make a `require` no-op and skip the overlay. Drop
|
||||
;; it from the loaded set so a require pulls the overlay from the source roots
|
||||
;; (like clojure.test); the primitives stay defined either way.
|
||||
(hashtable-delete! loaded-ns "clojure.core.async")
|
||||
|
||||
;; Does `name` already have vars in the var-table? A namespace baked into the
|
||||
;; image after the snapshot above — an AOT'd app namespace in a `jolt build`
|
||||
;; binary — exists in memory with no source file; a later `require` of it must
|
||||
|
|
|
|||
|
|
@ -74,7 +74,8 @@
|
|||
;; :refer :all — bring in every public var (require :refer :all)
|
||||
((and (keyword? v) (string=? (keyword-t-name v) "all"))
|
||||
(chez-register-refer-all! cns target))
|
||||
((pvec? v)
|
||||
;; :refer [a b] or :refer (a b) — both forms list names to bring in.
|
||||
((or (pvec? v) (cseq? v) (empty-list-t? v))
|
||||
(for-each (lambda (n)
|
||||
(when (symbol-t? n) (chez-register-refer! cns (symbol-t-name n) target)))
|
||||
(seq->list v))))))))
|
||||
|
|
|
|||
|
|
@ -249,10 +249,13 @@
|
|||
(jolt-persistent! (reduce-seq (lambda (t x) (jolt-conj! t x)) (jolt-transient-new to) (jolt-seq from)))))
|
||||
|
||||
(define (range-from n) (cseq-lazy n (lambda () (range-from (+ n 1)))))
|
||||
;; An empty range is () (jolt-empty-list), NOT nil — (range 0) and (range 5 5) are
|
||||
;; empty seqs in Clojure, so (= () (range 0)) holds. The same () terminates the
|
||||
;; lazy tail of a non-empty range (jolt-empty-list seqs back to nil, see jolt-take).
|
||||
(define (range-bounded n end step)
|
||||
(if (if (> step 0.0) (< n end) (> n end))
|
||||
(cseq-lazy n (lambda () (range-bounded (+ n step) end step)))
|
||||
jolt-nil))
|
||||
jolt-empty-list))
|
||||
;; numeric tower: exact 0/1 defaults so (range 3) yields exact ints
|
||||
;; (= JVM longs); flonum args still produce flonums (Scheme arithmetic preserves).
|
||||
(define jolt-range
|
||||
|
|
|
|||
667
stdlib/clojure/core/async.clj
Normal file
667
stdlib/clojure/core/async.clj
Normal file
|
|
@ -0,0 +1,667 @@
|
|||
;; clojure.core.async — higher-level dataflow API over the channel primitives.
|
||||
;;
|
||||
;; The primitives (chan, <!, >!, <!!, >!!, close!, put!, take!, offer!, timeout,
|
||||
;; promise-chan, buffer/dropping-buffer/sliding-buffer, go/go-loop/thread, go-spawn)
|
||||
;; are provided natively (host/chez/java/async.ss) on real OS threads. This overlay
|
||||
;; adds the portable dataflow operators — alts!, pipe, pipeline, split, reduce,
|
||||
;; transduce, mult, mix, pub/sub, map, merge, and the deprecated map</map>/… —
|
||||
;; ported from clojure.core.async over those primitives. Because go blocks are real
|
||||
;; threads, parking ops are ordinary blocking ops and work anywhere; this is a
|
||||
;; superset of the JVM model (no fixed thread pool, no pending-op limit).
|
||||
|
||||
(ns clojure.core.async
|
||||
(:refer-clojure :exclude [reduce transduce into merge map take partition partition-by]))
|
||||
|
||||
;; --- alts -------------------------------------------------------------------
|
||||
;; do-alts polls each port non-blockingly under its own channel lock; the first
|
||||
;; ready op wins. A take port is ready when a value (or closed nil) is available;
|
||||
;; a put spec [ch val] is ready when the value can be offered. Polls with a 1ms
|
||||
;; backoff (no cross-channel wait-set).
|
||||
|
||||
(defn- alt-attempt [port]
|
||||
(if (vector? port)
|
||||
(let [ch (nth port 0) v (nth port 1)]
|
||||
(assert (some? v) "Can't put nil on channel")
|
||||
(let [r (clojure.core.async/__offer! ch v)] ; true | false (closed) | nil (would block)
|
||||
(when (some? r) [r ch])))
|
||||
(let [r (clojure.core.async/__poll! port)] ; value | nil (closed) | ::none
|
||||
(when (not= r ::none) [r port]))))
|
||||
|
||||
(defn do-alts
|
||||
"Returns [val port] for the first ready op among ports. ports is a vector of
|
||||
take ports and/or [channel val] put specs. opts may include :priority true
|
||||
(try in order) and :default val (return [val :default] if none ready)."
|
||||
[ports opts]
|
||||
(assert (pos? (count ports)) "alts must have at least one channel operation")
|
||||
(let [ports (vec ports)
|
||||
n (count ports)
|
||||
priority (:priority opts)
|
||||
has-default (contains? opts :default)]
|
||||
;; Scan ports from a random start (sequential, wrapping) so a non-priority alts
|
||||
;; is fair without allocating a fresh shuffle every poll. With :priority the scan
|
||||
;; starts at 0 (declared order). Returns the first ready op.
|
||||
(loop [first? true]
|
||||
(let [start (if priority 0 (rand-int n))
|
||||
hit (loop [k 0]
|
||||
(when (< k n)
|
||||
(let [j (+ start k) i (if (< j n) j (- j n))]
|
||||
(or (alt-attempt (nth ports i))
|
||||
(recur (inc k))))))]
|
||||
(cond
|
||||
hit hit
|
||||
(and first? has-default) [(:default opts) :default]
|
||||
:else (do (Thread/sleep 1) (recur false)))))))
|
||||
|
||||
(defn alts!!
|
||||
"Completes at most one of several channel operations. ports is a vector of take
|
||||
ports and/or [channel val] put specs. Returns [val port]. Blocks until ready."
|
||||
[ports & {:as opts}]
|
||||
(do-alts ports opts))
|
||||
|
||||
(defn alts!
|
||||
"Like alts!!. In jolt a go block is a real thread, so parking and blocking alts
|
||||
are the same operation."
|
||||
[ports & {:as opts}]
|
||||
(do-alts ports opts))
|
||||
|
||||
(defn poll!
|
||||
"Takes a val from port if possible immediately. Never blocks. Returns the value
|
||||
or nil."
|
||||
[port]
|
||||
(let [r (clojure.core.async/__poll! port)]
|
||||
(when (not= r ::none) r)))
|
||||
|
||||
;; --- thread variants --------------------------------------------------------
|
||||
|
||||
(defn thread-call
|
||||
"Executes f in another thread, returning a channel that receives f's result then
|
||||
closes."
|
||||
([f] (clojure.core.async/go-spawn f))
|
||||
([f _workload] (clojure.core.async/go-spawn f)))
|
||||
|
||||
(defmacro io-thread
|
||||
"Executes body in another thread, returning a channel that receives the result
|
||||
then closes."
|
||||
[& body]
|
||||
`(thread-call (fn [] ~@body) :io))
|
||||
|
||||
;; --- pipe / pipeline --------------------------------------------------------
|
||||
|
||||
(defn pipe
|
||||
"Takes elements from the from channel and supplies them to the to channel.
|
||||
Closes to when from closes unless close? is false."
|
||||
([from to] (pipe from to true))
|
||||
([from to close?]
|
||||
(go-loop []
|
||||
(let [v (<! from)]
|
||||
(if (nil? v)
|
||||
(when close? (close! to))
|
||||
(when (>! to v)
|
||||
(recur)))))
|
||||
to))
|
||||
|
||||
(defn- pipeline*
|
||||
[n to xf from close? ex-handler type]
|
||||
(assert (pos? n))
|
||||
(let [jobs (chan n)
|
||||
results (chan n)
|
||||
process (fn [job]
|
||||
(if (nil? job)
|
||||
(do (close! results) nil)
|
||||
(let [v (nth job 0) p (nth job 1)
|
||||
res (chan 1 xf ex-handler)]
|
||||
(>!! res v)
|
||||
(close! res)
|
||||
(put! p res)
|
||||
true)))
|
||||
afn (fn [job]
|
||||
(if (nil? job)
|
||||
(do (close! results) nil)
|
||||
(let [v (nth job 0) p (nth job 1)
|
||||
res (chan 1)]
|
||||
(xf v res)
|
||||
(put! p res)
|
||||
true)))]
|
||||
(dotimes [_ n]
|
||||
(case type
|
||||
(:blocking :compute) (thread
|
||||
(loop []
|
||||
(let [job (<!! jobs)]
|
||||
(when (process job)
|
||||
(recur)))))
|
||||
:async (go-loop []
|
||||
(let [job (<! jobs)]
|
||||
(when (afn job)
|
||||
(recur))))))
|
||||
(go-loop []
|
||||
(let [v (<! from)]
|
||||
(if (nil? v)
|
||||
(close! jobs)
|
||||
(let [p (chan 1)]
|
||||
(>! jobs [v p])
|
||||
(>! results p)
|
||||
(recur)))))
|
||||
(go-loop []
|
||||
(let [p (<! results)]
|
||||
(if (nil? p)
|
||||
(when close? (close! to))
|
||||
(let [res (<! p)]
|
||||
(loop []
|
||||
(let [v (<! res)]
|
||||
(when (and (not (nil? v)) (>! to v))
|
||||
(recur))))
|
||||
(recur)))))))
|
||||
|
||||
(defn pipeline
|
||||
"Takes elements from from, applies transducer xf with parallelism n, supplies to
|
||||
to. Outputs are ordered relative to inputs."
|
||||
([n to xf from] (pipeline n to xf from true))
|
||||
([n to xf from close?] (pipeline n to xf from close? nil))
|
||||
([n to xf from close? ex-handler] (pipeline* n to xf from close? ex-handler :compute)))
|
||||
|
||||
(defn pipeline-blocking
|
||||
"Like pipeline, for blocking operations."
|
||||
([n to xf from] (pipeline-blocking n to xf from true))
|
||||
([n to xf from close?] (pipeline-blocking n to xf from close? nil))
|
||||
([n to xf from close? ex-handler] (pipeline* n to xf from close? ex-handler :blocking)))
|
||||
|
||||
(defn pipeline-async
|
||||
"Like pipeline, for async fns af of two args [input result-channel]."
|
||||
([n to af from] (pipeline-async n to af from true))
|
||||
([n to af from close?] (pipeline* n to af from close? nil :async)))
|
||||
|
||||
(defn split
|
||||
"Splits ch by predicate p into [true-chan false-chan]."
|
||||
([p ch] (split p ch nil nil))
|
||||
([p ch t-buf-or-n f-buf-or-n]
|
||||
(let [tc (chan t-buf-or-n)
|
||||
fc (chan f-buf-or-n)]
|
||||
(go-loop []
|
||||
(let [v (<! ch)]
|
||||
(if (nil? v)
|
||||
(do (close! tc) (close! fc))
|
||||
(when (>! (if (p v) tc fc) v)
|
||||
(recur)))))
|
||||
[tc fc])))
|
||||
|
||||
;; --- reduce / transduce / collection sinks ----------------------------------
|
||||
|
||||
(defn reduce
|
||||
"Returns a channel with the single result of reducing ch with f from init."
|
||||
[f init ch]
|
||||
(go-loop [ret init]
|
||||
(let [v (<! ch)]
|
||||
(if (nil? v)
|
||||
ret
|
||||
(let [ret' (f ret v)]
|
||||
(if (reduced? ret')
|
||||
@ret'
|
||||
(recur ret')))))))
|
||||
|
||||
(defn transduce
|
||||
"async/reduces ch with the transformation (xform f), returning a channel with the
|
||||
result."
|
||||
[xform f init ch]
|
||||
(let [f (xform f)]
|
||||
(go
|
||||
(let [ret (<! (reduce f init ch))]
|
||||
(f ret)))))
|
||||
|
||||
(defn- bounded-count [n coll]
|
||||
(if (counted? coll)
|
||||
(min n (count coll))
|
||||
(loop [i 0 s (seq coll)]
|
||||
(if (and s (< i n))
|
||||
(recur (inc i) (next s))
|
||||
i))))
|
||||
|
||||
(defn onto-chan!
|
||||
"Puts the contents of coll into ch, closing ch after unless close? is false.
|
||||
Returns a channel that closes when done."
|
||||
([ch coll] (onto-chan! ch coll true))
|
||||
([ch coll close?]
|
||||
(go-loop [vs (seq coll)]
|
||||
(if (and vs (>! ch (first vs)))
|
||||
(recur (next vs))
|
||||
(when close?
|
||||
(close! ch))))))
|
||||
|
||||
(defn to-chan!
|
||||
"Returns a channel containing the contents of coll, closing when exhausted."
|
||||
[coll]
|
||||
(let [c (bounded-count 100 coll)]
|
||||
(if (pos? c)
|
||||
(let [ch (chan c)]
|
||||
(onto-chan! ch coll)
|
||||
ch)
|
||||
(let [ch (chan)]
|
||||
(close! ch)
|
||||
ch))))
|
||||
|
||||
(defn onto-chan!!
|
||||
"Like onto-chan! for use when accessing coll might block."
|
||||
([ch coll] (onto-chan!! ch coll true))
|
||||
([ch coll close?]
|
||||
(thread
|
||||
(loop [vs (seq coll)]
|
||||
(if (and vs (>!! ch (first vs)))
|
||||
(recur (next vs))
|
||||
(when close?
|
||||
(close! ch)))))))
|
||||
|
||||
(defn to-chan!!
|
||||
"Like to-chan! for use when accessing coll might block."
|
||||
[coll]
|
||||
(let [c (bounded-count 100 coll)]
|
||||
(if (pos? c)
|
||||
(let [ch (chan c)]
|
||||
(onto-chan!! ch coll)
|
||||
ch)
|
||||
(let [ch (chan)]
|
||||
(close! ch)
|
||||
ch))))
|
||||
|
||||
(defn onto-chan
|
||||
"Deprecated - use onto-chan! or onto-chan!!"
|
||||
([ch coll] (onto-chan! ch coll true))
|
||||
([ch coll close?] (onto-chan! ch coll close?)))
|
||||
|
||||
(defn to-chan
|
||||
"Deprecated - use to-chan! or to-chan!!"
|
||||
[coll]
|
||||
(to-chan! coll))
|
||||
|
||||
(defn into
|
||||
"Returns a channel with the single collection result of conjoining items from ch
|
||||
onto coll. ch must close first."
|
||||
[coll ch]
|
||||
(reduce conj coll ch))
|
||||
|
||||
(defn take
|
||||
"Returns a channel that returns at most n items from ch, then closes."
|
||||
([n ch] (take n ch nil))
|
||||
([n ch buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(go (loop [x 0]
|
||||
(when (< x n)
|
||||
(let [v (<! ch)]
|
||||
(when (not (nil? v))
|
||||
(>! out v)
|
||||
(recur (inc x))))))
|
||||
(close! out))
|
||||
out)))
|
||||
|
||||
;; --- mult / tap -------------------------------------------------------------
|
||||
|
||||
(defprotocol Mux
|
||||
(muxch* [_]))
|
||||
|
||||
(defprotocol Mult
|
||||
(tap* [m ch close?])
|
||||
(untap* [m ch])
|
||||
(untap-all* [m]))
|
||||
|
||||
(defn mult
|
||||
"Creates a mult of ch. Copies can be created with tap and removed with untap.
|
||||
Each item is distributed to all taps synchronously."
|
||||
[ch]
|
||||
(let [cs (atom {})
|
||||
m (reify
|
||||
Mux
|
||||
(muxch* [_] ch)
|
||||
Mult
|
||||
(tap* [_ ch close?] (swap! cs assoc ch close?) nil)
|
||||
(untap* [_ ch] (swap! cs dissoc ch) nil)
|
||||
(untap-all* [_] (reset! cs {}) nil))
|
||||
dchan (chan 1)
|
||||
dctr (atom nil)
|
||||
done (fn [_] (when (zero? (swap! dctr dec))
|
||||
(put! dchan true)))]
|
||||
(go-loop []
|
||||
(let [val (<! ch)]
|
||||
(if (nil? val)
|
||||
(doseq [[c close?] @cs]
|
||||
(when close? (close! c)))
|
||||
(let [chs (keys @cs)]
|
||||
(reset! dctr (count chs))
|
||||
(doseq [c chs]
|
||||
(when-not (put! c val done)
|
||||
(untap* m c)))
|
||||
(when (seq chs)
|
||||
(<! dchan))
|
||||
(recur)))))
|
||||
m))
|
||||
|
||||
(defn tap
|
||||
"Copies the mult source onto ch. Closes ch when the source closes unless close?
|
||||
is false."
|
||||
([mult ch] (tap mult ch true))
|
||||
([mult ch close?] (tap* mult ch close?) ch))
|
||||
|
||||
(defn untap
|
||||
"Disconnects ch from a mult."
|
||||
[mult ch]
|
||||
(untap* mult ch))
|
||||
|
||||
(defn untap-all
|
||||
"Disconnects all channels from a mult."
|
||||
[mult]
|
||||
(untap-all* mult))
|
||||
|
||||
;; --- mix --------------------------------------------------------------------
|
||||
|
||||
(defprotocol Mix
|
||||
(admix* [m ch])
|
||||
(unmix* [m ch])
|
||||
(unmix-all* [m])
|
||||
(toggle* [m state-map])
|
||||
(solo-mode* [m mode]))
|
||||
|
||||
(defn mix
|
||||
"Creates a mix of input channels put onto out. Inputs are added with admix,
|
||||
removed with unmix, and toggled (:mute/:pause/:solo) with toggle."
|
||||
[out]
|
||||
(let [cs (atom {})
|
||||
solo-modes #{:mute :pause}
|
||||
solo-mode (atom :mute)
|
||||
change (chan (sliding-buffer 1))
|
||||
changed #(put! change true)
|
||||
pick (fn [attr chs]
|
||||
(reduce-kv
|
||||
(fn [ret c v]
|
||||
(if (attr v) (conj ret c) ret))
|
||||
#{} chs))
|
||||
calc-state (fn []
|
||||
(let [chs @cs
|
||||
mode @solo-mode
|
||||
solos (pick :solo chs)
|
||||
pauses (pick :pause chs)]
|
||||
{:solos solos
|
||||
:mutes (pick :mute chs)
|
||||
:reads (conj
|
||||
(if (and (= mode :pause) (seq solos))
|
||||
(vec solos)
|
||||
(vec (remove pauses (keys chs))))
|
||||
change)}))
|
||||
m (reify
|
||||
Mux
|
||||
(muxch* [_] out)
|
||||
Mix
|
||||
(admix* [_ ch] (swap! cs assoc ch {}) (changed))
|
||||
(unmix* [_ ch] (swap! cs dissoc ch) (changed))
|
||||
(unmix-all* [_] (reset! cs {}) (changed))
|
||||
(toggle* [_ state-map] (swap! cs (partial merge-with clojure.core/merge) state-map) (changed))
|
||||
(solo-mode* [_ mode]
|
||||
(assert (solo-modes mode) (str "mode must be one of: " solo-modes))
|
||||
(reset! solo-mode mode)
|
||||
(changed)))]
|
||||
(go-loop [state (calc-state)]
|
||||
(let [{:keys [solos mutes reads]} state
|
||||
[v c] (alts! reads)]
|
||||
(if (or (nil? v) (= c change))
|
||||
(do (when (nil? v)
|
||||
(swap! cs dissoc c))
|
||||
(recur (calc-state)))
|
||||
(if (or (solos c)
|
||||
(and (empty? solos) (not (mutes c))))
|
||||
(when (>! out v)
|
||||
(recur state))
|
||||
(recur state)))))
|
||||
m))
|
||||
|
||||
(defn admix
|
||||
"Adds ch as an input to the mix."
|
||||
[mix ch]
|
||||
(admix* mix ch))
|
||||
|
||||
(defn unmix
|
||||
"Removes ch as an input to the mix."
|
||||
[mix ch]
|
||||
(unmix* mix ch))
|
||||
|
||||
(defn unmix-all
|
||||
"Removes all inputs from the mix."
|
||||
[mix]
|
||||
(unmix-all* mix))
|
||||
|
||||
(defn toggle
|
||||
"Atomically sets the state of one or more channels in a mix."
|
||||
[mix state-map]
|
||||
(toggle* mix state-map))
|
||||
|
||||
(defn solo-mode
|
||||
"Sets the solo mode of the mix (:mute or :pause)."
|
||||
[mix mode]
|
||||
(solo-mode* mix mode))
|
||||
|
||||
;; --- pub / sub --------------------------------------------------------------
|
||||
|
||||
(defprotocol Pub
|
||||
(sub* [p v ch close?])
|
||||
(unsub* [p v ch])
|
||||
(unsub-all* [p] [p v]))
|
||||
|
||||
(defn pub
|
||||
"Creates a pub of ch partitioned by topic-fn. Subscribe with sub."
|
||||
([ch topic-fn] (pub ch topic-fn (constantly nil)))
|
||||
([ch topic-fn buf-fn]
|
||||
(let [mults (atom {})
|
||||
ensure-mult (fn [topic]
|
||||
(or (get @mults topic)
|
||||
(get (swap! mults
|
||||
#(if (% topic) % (assoc % topic (mult (chan (buf-fn topic))))))
|
||||
topic)))
|
||||
p (reify
|
||||
Mux
|
||||
(muxch* [_] ch)
|
||||
Pub
|
||||
(sub* [_p topic ch close?]
|
||||
(let [m (ensure-mult topic)]
|
||||
(tap m ch close?)))
|
||||
(unsub* [_p topic ch]
|
||||
(when-let [m (get @mults topic)]
|
||||
(untap m ch)))
|
||||
(unsub-all* [_] (reset! mults {}))
|
||||
(unsub-all* [_ topic] (swap! mults dissoc topic)))]
|
||||
(go-loop []
|
||||
(let [val (<! ch)]
|
||||
(if (nil? val)
|
||||
(doseq [m (vals @mults)]
|
||||
(close! (muxch* m)))
|
||||
(let [topic (topic-fn val)
|
||||
m (get @mults topic)]
|
||||
(when m
|
||||
(when-not (>! (muxch* m) val)
|
||||
(swap! mults dissoc topic)))
|
||||
(recur)))))
|
||||
p)))
|
||||
|
||||
(defn sub
|
||||
"Subscribes ch to a topic of pub p."
|
||||
([p topic ch] (sub p topic ch true))
|
||||
([p topic ch close?] (sub* p topic ch close?)))
|
||||
|
||||
(defn unsub
|
||||
"Unsubscribes ch from a topic of pub p."
|
||||
[p topic ch]
|
||||
(unsub* p topic ch))
|
||||
|
||||
(defn unsub-all
|
||||
"Unsubscribes all channels from a pub, or from a topic."
|
||||
([p] (unsub-all* p))
|
||||
([p topic] (unsub-all* p topic)))
|
||||
|
||||
;; --- map / merge ------------------------------------------------------------
|
||||
|
||||
(defn map
|
||||
"Applies f to the set of first items from each source channel, then second, etc.
|
||||
Closes the output channel when any source closes."
|
||||
([f chs] (map f chs nil))
|
||||
([f chs buf-or-n]
|
||||
(let [chs (vec chs)
|
||||
out (chan buf-or-n)
|
||||
cnt (count chs)
|
||||
rets (atom (vec (repeat cnt nil)))
|
||||
dchan (chan 1)
|
||||
dctr (atom nil)
|
||||
done (mapv (fn [i]
|
||||
(fn [ret]
|
||||
(swap! rets assoc i ret)
|
||||
(when (zero? (swap! dctr dec))
|
||||
(put! dchan @rets))))
|
||||
(range cnt))]
|
||||
(if (zero? cnt)
|
||||
(close! out)
|
||||
(go-loop []
|
||||
(reset! dctr cnt)
|
||||
(dotimes [i cnt]
|
||||
(take! (nth chs i) (nth done i)))
|
||||
(let [rets (<! dchan)]
|
||||
(if (some nil? rets)
|
||||
(close! out)
|
||||
(do (>! out (apply f rets))
|
||||
(recur))))))
|
||||
out)))
|
||||
|
||||
(defn merge
|
||||
"Returns a channel with all values taken from the source channels chs. Closes
|
||||
after all sources close."
|
||||
([chs] (merge chs nil))
|
||||
([chs buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(go-loop [cs (vec chs)]
|
||||
(if (pos? (count cs))
|
||||
(let [[v c] (alts! cs)]
|
||||
(if (nil? v)
|
||||
(recur (filterv #(not= c %) cs))
|
||||
(do (>! out v)
|
||||
(recur cs))))
|
||||
(close! out)))
|
||||
out)))
|
||||
|
||||
;; --- deprecated channel ops (rewritten as go-loops) -------------------------
|
||||
|
||||
(defn map<
|
||||
"Deprecated - use a transducer. Returns a read-side channel mapping f over ch."
|
||||
[f ch]
|
||||
(let [out (chan)]
|
||||
(go-loop []
|
||||
(let [v (<! ch)]
|
||||
(if (nil? v) (close! out) (do (>! out (f v)) (recur)))))
|
||||
out))
|
||||
|
||||
(defn map>
|
||||
"Deprecated - use a transducer. Returns a write-side channel mapping f into out."
|
||||
[f out]
|
||||
(let [in (chan)]
|
||||
(go-loop []
|
||||
(let [v (<! in)]
|
||||
(if (nil? v) (close! out) (do (>! out (f v)) (recur)))))
|
||||
in))
|
||||
|
||||
(defn filter<
|
||||
"Deprecated - use a transducer."
|
||||
([p ch] (filter< p ch nil))
|
||||
([p ch buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(go-loop []
|
||||
(let [val (<! ch)]
|
||||
(if (nil? val)
|
||||
(close! out)
|
||||
(do (when (p val) (>! out val))
|
||||
(recur)))))
|
||||
out)))
|
||||
|
||||
(defn remove<
|
||||
"Deprecated - use a transducer."
|
||||
([p ch] (remove< p ch nil))
|
||||
([p ch buf-or-n] (filter< (complement p) ch buf-or-n)))
|
||||
|
||||
(defn filter>
|
||||
"Deprecated - use a transducer."
|
||||
[p out]
|
||||
(let [in (chan)]
|
||||
(go-loop []
|
||||
(let [v (<! in)]
|
||||
(if (nil? v)
|
||||
(close! out)
|
||||
(do (when (p v) (>! out v))
|
||||
(recur)))))
|
||||
in))
|
||||
|
||||
(defn remove>
|
||||
"Deprecated - use a transducer."
|
||||
[p out]
|
||||
(filter> (complement p) out))
|
||||
|
||||
(defn- mapcat* [f in out]
|
||||
(go-loop []
|
||||
(let [val (<! in)]
|
||||
(if (nil? val)
|
||||
(close! out)
|
||||
(do (doseq [v (f val)]
|
||||
(>! out v))
|
||||
(recur))))))
|
||||
|
||||
(defn mapcat<
|
||||
"Deprecated - use a transducer."
|
||||
([f in] (mapcat< f in nil))
|
||||
([f in buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(mapcat* f in out)
|
||||
out)))
|
||||
|
||||
(defn mapcat>
|
||||
"Deprecated - use a transducer."
|
||||
([f out] (mapcat> f out nil))
|
||||
([f out buf-or-n]
|
||||
(let [in (chan buf-or-n)]
|
||||
(mapcat* f in out)
|
||||
in)))
|
||||
|
||||
(defn unique
|
||||
"Deprecated - use a transducer. Drops consecutive duplicates."
|
||||
([ch] (unique ch nil))
|
||||
([ch buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(go (loop [last nil]
|
||||
(let [v (<! ch)]
|
||||
(when (not (nil? v))
|
||||
(if (= v last)
|
||||
(recur last)
|
||||
(do (>! out v)
|
||||
(recur v))))))
|
||||
(close! out))
|
||||
out)))
|
||||
|
||||
(defn partition
|
||||
"Deprecated - use a transducer. Partitions ch into vectors of n."
|
||||
([n ch] (partition n ch nil))
|
||||
([n ch buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(go-loop [arr [] idx 0]
|
||||
(let [v (<! ch)]
|
||||
(if (not (nil? v))
|
||||
(let [arr (conj arr v) new-idx (inc idx)]
|
||||
(if (< new-idx n)
|
||||
(recur arr new-idx)
|
||||
(do (>! out arr) (recur [] 0))))
|
||||
(do (when (> idx 0) (>! out arr))
|
||||
(close! out)))))
|
||||
out)))
|
||||
|
||||
(defn partition-by
|
||||
"Deprecated - use a transducer. Partitions ch by runs of (f v)."
|
||||
([f ch] (partition-by f ch nil))
|
||||
([f ch buf-or-n]
|
||||
(let [out (chan buf-or-n)]
|
||||
(go-loop [lst [] last ::nothing]
|
||||
(let [v (<! ch)]
|
||||
(if (not (nil? v))
|
||||
(let [new-itm (f v)]
|
||||
(if (or (= new-itm last) (identical? last ::nothing))
|
||||
(recur (conj lst v) new-itm)
|
||||
(do (>! out lst) (recur [v] new-itm))))
|
||||
(do (when (> (count lst) 0) (>! out lst))
|
||||
(close! out)))))
|
||||
out)))
|
||||
34
stdlib/clojure/core/async/lab.clj
Normal file
34
stdlib/clojure/core/async/lab.clj
Normal file
|
|
@ -0,0 +1,34 @@
|
|||
;; 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))
|
||||
|
|
@ -3349,4 +3349,8 @@
|
|||
{:suite "symbols / interning" :label "equal symbols share an interned name string" :expected "true" :actual "(let [a (quote ?foo) b (quote ?foo)] (identical? (name a) (name b)))"}
|
||||
{:suite "reader / unquote" :label "~x reads as clojure.core/unquote" :expected "true" :actual "(= (quote (clojure.core/unquote v)) (read-string \"~v\"))"}
|
||||
{:suite "reader / unquote" :label "~@x reads as clojure.core/unquote-splicing" :expected "true" :actual "(= (quote clojure.core/unquote-splicing) (first (read-string \"~@xs\")))"}
|
||||
{:suite "clojure.core / range" :label "(range 0) is the empty seq ()" :expected "true" :actual "(= () (range 0))"}
|
||||
{:suite "clojure.core / range" :label "(range 5 5) is the empty seq ()" :expected "true" :actual "(= () (range 5 5))"}
|
||||
{:suite "clojure.core / range" :label "empty range is not nil" :expected "true" :actual "(some? (range 5 0))"}
|
||||
{:suite "clojure.core / range" :label "empty range count" :expected "0" :actual "(count (range 0))"}
|
||||
]
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue