 (ns backtype.storm.daemon.acker
   (:import [backtype.storm.task OutputCollector TopologyContext IBolt])
   (:import [backtype.storm.tuple Tuple Fields])
-  (:import [backtype.storm.utils TimeCacheMap])
+  (:import [backtype.storm.utils TimeCacheMap MutableObject])
   (:import [java.util List Map])
   (:use [backtype.storm config util log])
   (:gen-class

                 (cond (= 0 (:val curr))
                       (do
                         (.remove pending id)
                         (acker-emit-direct @output-collector
                                            (:spout-task curr)
                                            ACKER-ACK-STREAM-ID
                                            [id]

  )

(defn mk-acker-bolt []
  (let [output-collector (MutableObject.)
        pending (MutableObject.)]
    (reify IBolt
      (^void prepare [this ^Map storm-conf ^TopologyContext context ^OutputCollector collector]
               (.setObject output-collector collector)
               (.setObject pending (TimeCacheMap. (.maxTopologyMessageTimeout context)))
               )
      (^void execute [this ^Tuple tuple]
             (let [id (.getValue tuple 0)
                   ^TimeCacheMap pending (.getObject pending)
                   ^OutputCollector output-collector (.getObject output-collector)
                   curr (.get pending id)
                   curr (condp = (.getSourceStreamId tuple)
                            ACKER-INIT-STREAM-ID (-> curr
