                  (reset! task nil))
                ; We're still valid; keep going
                (f)))]
-               
+
      (fn stream [event]
        ; Bump the time we're allowed to keep running for.
        (if (and (:ttl event) (:time event))

  [args])

(defn dual
  "A stream which splits events into two mirror-images streams, based on (pred
  e). If (pred e) is true, calls (true-stream e) and (false-stream (expire e)).
  If (pred e) is false, does the opposite. Expired events are forwarded to both
  streams.
  
  (pred e) is always called once per incoming event."
  [pred true-stream false-stream]
  (fn stream [event]

              (when-not (zero? dt)
                (let [diff (/ (- m (:metric prev-event)) dt)]
                  (call-rescue (assoc event :metric diff) children))))))))))

(defn ddt
  "Differentiate metrics with respect to time. With no args, emits an event for
