 (defn under
   "Passes on events only when their metric is smaller than x"
   [x & children]
-  (fn [event]
+  (fn stream [event]
     (when-let [m (:metric event)]
       (when (> x m)
         (call-rescue event children)))))


  (let [start-time (ref nil)
        buffer     (ref [])]
    (fn [event]
      (let [windows (dosync
                      (cond
                        ; No time

        ; A list of all bins we're tracking.
        bins (ref {})
        ; Eventually finish a bin.
        finish-eventually (fn finish-eventually [bin start]
          (.start (new Thread (bound-fn thread []
                         (let [end (+ start interval)]
                           ; Sleep until this bin is past
                           (Thread/sleep (max 0 (* 1000 (- end (unix-time)))))
