prpr-stream.clj

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

(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))

;; @(stream/take! @result)

(def r (ddo [u-s (stream.cross/set-streams-union {:default-key-fn :id
                                                  :skey-streams {:ss1 ss1
                                                                 :ss2 ss2
                                                                 :ss3 ss3}})]
         (->> u-s
              (stream/map
               (fn [r]
                 (prn r)
                 r))
              (prpr.stream/count-all-throw
               "count results"))))