   (fresh? [_] false))
 
 (defprotocol SignalProtocol
-  (sources [s]))
+  (sources [s])
+  (message-emitter [s]))
 
 (defn signal?
   [x]
   (satisfies? SignalProtocol x))
 
+(defn- subscribed-topics-message-emitter
+  [topics]
+  (let [topics (if (set? topics) topics (set topics))]
+    (fn [prev event]
+      (if (contains? topics (:topic event))
+        (->Fresh (:value event))
+        (->Cached (:value prev))))))
+
 (defrecord Signal
-  [init message-emitter event-emitter]
+  [init message-emitter subscribed-topics event-emitter external-event-source]
   SignalProtocol
   (sources [_]
     (into #{}
           (comp (filter identity)
                 (mapcat (comp flatten vector first))
                 (filter signal?))
-          [message-emitter event-emitter])))
-
-(defn- event-relay
+          [message-emitter event-emitter]))
+  (message-emitter [_]
+    (cond
+      message-emitter message-emitter
+      subscribed-topics [:events (subscribed-topics-message-emitter subscribed-topics)])))
+
+(defn- messages-from-events
   [topic]
   (fn [prev event]
     (if (= topic (:topic event))


(defn- build-message-mult
  [signal mult-map]
  (if-let [[tmpl msg-fn] (:message-emitter signal)]
    (let [c-in (tap-template tmpl mult-map)
          c-out (chan)]
      (spawn-message-loop! (:init signal) msg-fn c-in c-out)

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

(defn constant
  [x]
