(define tbatch
  (case-lambda
    ((reducer)
     (tbatch null-transducer reducer))
    ((t r)
     (lambda (reducer)
       (let ((cur-reducer (t r))
             (cur-state (r)))
         (case-lambda
           (() (reducer))
           ((result)
            ;; Check whether we still have accumulated state
            (if (equal? cur-state (cur-reducer))
                (reducer result)
                ;; Pass the accumulated state downstream and handle
                ;; whether the downstream accepts new values.
                (let ((new-res (reducer result (cur-reducer cur-state))))
                  (if (reduced? new-res)
                      (reducer (unreduce new-res))
                      (reducer new-res)))))
           ((result value)
            (let ((val (cur-reducer cur-state value)))
              (cond
               ;; cur-reducer is done. Push value downstream
               ;; re-instantiate the state and the cur-reducer
               ((reduced? val)
                (let ((unreduced-val (unreduce val)))
                  (set! cur-reducer (t r))
                  (set! cur-state (cur-reducer))
                  (reducer result (cur-reducer unreduced-val))))
               (else
                (set! cur-state val)
                result))))))))))