prpr-stream.clj

(require '[clojure.core.async :refer [to-chan]])
(require '[manifold.stream :as stream])
(require '[prpr.stream.cross :as stream.cross])

(def c1 (to-chan [{:id 2 :value "a"} {:id 3} {:id 4}]))
(def c2 (to-chan [{:id 1} {:id 2 :value "b"} {:id 3}]))
(def c3 (to-chan [{:id 0} {:id 1} {:id 2 :value "c"}]))

(def cs1 (stream/->source c1))
(def cs2 (stream/->source c2))
(def cs3 (stream/->source c3))

(def ss1 (stream.cross/event-source->sorted-stream :id cs1))
(def ss2 (stream.cross/event-source->sorted-stream :id cs2))
(def ss3 (stream.cross/event-source->sorted-stream :id cs3))

(def result (stream.cross/set-streams-union {:default-key-fn :id
                                             :skey-streams {:ss1 ss1
                                                            :ss2 ss2
                                                            :ss3 ss3}}))

@(stream/take! @result)