                                                        (not= (:value msg) (:value prev)))
                                               msg))}}))
 
-
-(defn reducep
-  ([f source] (reducep f (f) source))
-  ([f init source]
-   (->> source
-        (foldp (fn [val acc] (f acc val)) init)
-        drop-repeats)))
-
-(defn transducep
-  ([xform f source] (reducep (xform f) (f) source))
-  ([xform f init source]
-   (reducep (xform f) init source)))
-
 (defn async
+  "Returns an \"asynchronous\" version of `source`, splitting off a new subgraph which
+  does not maintain consistent event ordering relative to the main graph. In exchange,
+  signals which depend on an `async` signal don't have to wait for the `source` to finish
+  computing new values. This function is mainly useful in multithreaded environments when
+  you don't want a slow computation to block the whole graph."
   [source]
   (let [topic source
         msgs->events (comp cat

                                              (->Fresh (:value value-msg))))}}))

(defn count
  [sig]
  (foldp #(inc %2) 0 sig))

(defn count-if
  [pred sig]
  (foldp (fn [v c]
           (if (pred v) (inc c) c))

                  :message-emitter (event-relay #{topic})})))

(defn mergeseq
  "Takes a sequence of signals `sigs`, and returns a new signal which relays fresh
  values from all of the source signals. When more than one source has fresh values
  at the same time, the first (leftmost) signal in `sigs` will take precedence and
  the other values will be discarded. The initial value of the returned signal is
  equal to the initial value of the first source signal."
  [sigs]
  (map->Signal {:init (:init (first sigs))
                :message-emitter {:sources sigs
