            (let [tuple (.deserialize deserializer ser-msg)
                  id (.getValue tuple 0)
                  [spout-id tuple-finished-info start-time-ms] (.remove pending id)
-                 time-delta (if start-time-ms (time-delta-ms start-time-ms))]
+                 time-delta (time-delta-ms start-time-ms)]
              (when spout-id
                (condp = (.getSourceStreamId tuple)
                  ACKER-ACK-STREAM-ID (.add event-queue #(ack-spout-msg spout storm-conf spout-id
-                                                                       tuple-finished-info time-delta task-stats))
+                                                                       tuple-finished-info time-delta task-stats sampler))
                  ACKER-FAIL-STREAM-ID (.add event-queue #(fail-spout-msg spout storm-conf spout-id
-                                                                         tuple-finished-info time-delta task-stats))
+                                                                         tuple-finished-info time-delta task-stats sampler))
                  )))
            ;; TODO: on failure, emit tuple to failure stream
            )))

                             (do
                               (.put pending root-id [message-id
                                                      {:stream out-stream-id :values values}
                                                      (if (sampler) (System/currentTimeMillis))])
                               (send-unanchored topology-context tasks-fn transfer-fn
                                                ACKER-INIT-STREAM-ID
                                                [root-id (bit-xor-vals out-ids) task-id]))
                             (when message-id
                               (.add event-queue #(ack-spout-msg spout storm-conf message-id {:stream out-stream-id :values values} nil task-stats))))
                           (or out-tasks [])
                           ))
        output-collector (reify ISpoutOutputCollector

             ;; TODO: how to handle incremental updates as well as synchronizations at same time
             ;; TODO: need to version tuples somehow
             (log-debug "Received tuple " tuple " at task " (.getThisTaskId topology-context))
             (.put tuple-start-times tuple (System/currentTimeMillis))
             
             (.execute bolt tuple)
             ))))]
