 
 (defn async
   [source]
-  (let [topic source]
+  (let [topic source
+        msgs->events (comp (filter fresh?)
+                           (map (fn [msg]
+                                  (->Event topic (:value msg)))))
+        events-channel-fn (fn [live-graph _]
+                            (async/tap (get (:mult-map live-graph) source)
+                                       (chan 1 msgs->events)))]
     (map->Signal {:init (:init source)
-                  :event-emitter [source (fn [message]
-                                           (when (fresh? message)
-                                             (:value message)))]
-                  :subscribed-topics #{topic}})))
+                  :deps #{source}
+                  :event-sources {topic events-channel-fn}
+                  :relayed-events #{topic}})))
 
 (defn mergeseq
   [sigs]

  [x]
  (satisfies? SignalProtocol x))

(defn- subscribed-topics-message-emitter
  [topics]
  (let [topics (if (set? topics) topics (set topics))]
    (fn [prev event]

  ([init] (input init (gen-topic)))
  ([init topic]
   (map->Signal {:init init
                 :relayed-events #{topic}})))

(defn constant
  [x]
