       (dosync
         (add (deref current) event)))))
 
-; Take the sum of every event over interval seconds and divide by the interval
-; size.
 (defn rate [interval & children]
+  "Take the sum of every event over interval seconds and divide by the interval
+  size."
   (part-time-fast interval
       (fn [] {:count (ref 0)
               :state (ref {})})

  (:use [clojure.contrib.math])
  (:use [clojure.contrib.logging]))

; Call each fn, in order, with event. Rescues and logs any failure.
(defmacro call-rescue [event children]
  `(do
     (doseq [child# ~children]
       (try

    (let [e (if (nil? v) (dissoc event k) (assoc event k v))]
      (call-rescue e children))))

(defmacro by [field & children]
  "Splits stream by field.
  Every time an event arrives with a new value of field, this macro invokes
  its enclosed form to return a *new*, distinct stream for that particular
  value."
  ; new-fork is a function which gives us a new copy of our children.
  ; table is a reference which maps (field event) to a fork (or list of
  ; children).
