    (let [last-event (ref nil)
          fill (bound-fn fill []
                 (call-rescue (merge @last-event update {:time (unix-time)}) children))
-         new-deferrable (fn [] (every! interval interval fill))
+         new-deferrable (fn new-deferrable []
+                          (every! interval interval fill))
          deferrable (ref nil)]
-     (fn [event]
+     (fn stream [event]
        ; Record last event
        (dosync (ref-set last-event event))
 

  (let [state (atom (list nil))  ; Events at t3, t2, and t1.
        swap (bound-fn swap []
               (let [[_ e2 e1] (swap! state
                                      (fn [[e3 e2 e1 :as state]]
                                        ; If no events have come in this
                                        ; interval, we preserve the last event
                                        ; in both places, which means we emit

(defn append
  "Conj events onto the given reference"
  [reference]
  (fn stream [event]
    (dosync
      (alter reference conj event))))

(defn register
  "Set reference to the most recent event that passes through."
  [reference]
  (fn stream [event]
    (dosync (ref-set reference event))))

(defn forward
  "Sends an event through a client"
  [client]
  (fn stream [event]
    (riemann.client/send-event client event)))

(defn match
