clojure.core.async/pipe

View this page on ClojureDocs

Type: function Added: Clojure unknown Examples: 3 Runnable: 0

([from to] [from to close?])

Takes elements from the from channel and supplies them to the to channel. By default, the to channel will be closed when the from channel closes, but can be determined by the close? parameter. Will stop consuming the from channel if the to channel closes

Examples

by runexec on 2014-10-01 . May have evaluation errors. ▶

(def cx (chan 1)) #'user/cx (def cy (chan 1)) #'user/cy (pipe cx cy) #<ManyToManyChannel clojure.core.async.impl.channels.ManyToManyChannel@5064227c> (put! cx "Going into CX") true (<!! cy) "Going into CX"

by beoliver on 2017-10-29 . May have evaluation errors. ▶

(def ch-1 (async/chan 10)) #'user/ch-1 (def ch-2 (async/chan 10)) #'user/ch-2 ;; start a thread that takes from channel 1, sleeping after each take (async/thread (loop [] (when-let [v (async/<!! ch-1)] (println "ch-1 taker got " v) (Thread/sleep 1000) (recur)))) #object[clojure.core.async.impl.channels.ManyToManyChannel ...] ;; start a thread that takes from channel 2 that does not sleep (async/thread (loop [] (when-let [v (async/<!! ch-2)] (println "ch-2 taker got " v) (recur)))) #object[clojure.core.async.impl.channels.ManyToManyChannel ...] ;; create a pipe from ch-1 to ch-2 (async/pipe ch-1 ch-2) #object[clojure.core.async.impl.channels.ManyToManyChannel ...] ;; lets try to add a new listener to channel 1. (async/thread (loop [] (when-let [v (async/<!! ch-1)] (println "second ch-1 taker got " v) (Thread/sleep 1000) (recur)))) #object[clojure.core.async.impl.channels.ManyToManyChannel ...] ;; put onto ch-1 and sleep after each put... we are interested in how the ;; pipe works... (async/thread (doseq [x (range 10)] (async/>!! ch-1 x) (Thread/sleep 500))) #object[clojure.core.async.impl.channels.ManyToManyChannel ...] ch-1 taker got 0 ch-2 taker got 1 second ch-1 taker got 2 ch-2 taker got 3 ch-1 taker got 4 ch-2 taker got 5 second ch-1 taker got 6 ch-2 taker got 7 ch-1 taker got 8 ch-2 taker got 9 ;; so a pipe is just a listener that tries to take from the from channel ;; and put onto the to channel, a pipe does dot restrict the ability to ;; *take* from the *from* channel.

by ferdinand-beyer on 2021-07-26 . May have evaluation errors. ▶

;;;; pipe is especially useful in combination with channel transducers, ;;;; to transform items in a channel. ;; Some input channel. (def input (to-chan (range 10))) ;; Create a channel with all items from input transformed. Here, we ;; create a transducer that multiplies by 5. ;; Note that pipe returns the to channel. (def transformed (pipe input (chan 1 (map (partial * 5))))) ;; Read transformed channel into a vector. (as/<!! (as/into [] transformed)) ; => [0 5 10 15 20 25 30 35 40 45]

Note by jayzawrotny

Returns the to channel argument.

---

Content from the matching ClojureDocs page, with authors credited on each contribution.