refactor acker -- doesn't need to be serializable

Nathan Marz · Nathan Marz · commit f59256607e8b · 2011-11-22T13:48:31.000-05:00
diff --git a/src/clj/backtype/storm/daemon/acker.clj b/src/clj/backtype/storm/daemon/acker.clj
@@ -1,14 +1,10 @@
 (ns backtype.storm.daemon.acker
-  (:import [backtype.storm.task OutputCollector IBolt TopologyContext])
+  (:import [backtype.storm.task OutputCollector TopologyContext])
   (:import [backtype.storm.tuple Tuple Fields])
   (:import [backtype.storm.utils TimeCacheMap])
+  (:import [backtype.storm.topology IRichBolt])
   (:import [java.util List Map])
-  (:use [backtype.storm config])
-  (:gen-class
-   :init init
-   :implements [backtype.storm.topology.IRichBolt]
-   :constructors {[] []}
-   :state state ))
+  (:use [backtype.storm config]))
 
 (def ACKER-COMPONENT-ID "__acker")
 (def ACKER-INIT-STREAM-ID "__ack_init")
@@ -24,11 +20,10 @@
   (.emitDirect collector task stream values)
   )
 
-;;TODO: this approach doesn't work... need to set it during prepare
 (defn mk-acker-bolt []
   (let [output-collector (atom nil)
         pending (atom nil)]
-    (reify IBolt
+    (reify IRichBolt
       (^void prepare [this ^Map storm-conf ^TopologyContext context ^OutputCollector collector]
                (reset! output-collector collector)
                (reset! pending (TimeCacheMap. (int (storm-conf TOPOLOGY-MESSAGE-TIMEOUT-SECS))))
@@ -67,23 +62,7 @@
                ))
       (^void cleanup [this]
              )
+      (declareOutputFields [this declarer]
+        (.declareStream declarer ACKER-ACK-STREAM-ID true (Fields. ["id"]))
+        (.declareStream declarer ACKER-FAIL-STREAM-ID true (Fields. ["id"])))
       )))
-
-(defn -init []
-  [[] (mk-acker-bolt)]
-  )
-  
-(defn -prepare [this ^Map storm-conf ^TopologyContext context ^OutputCollector collector]
-  (.prepare ^IBolt (. this state) storm-conf context collector))
-
-(defn -execute [this ^Tuple tuple]
-  (.execute ^IBolt (. this state) tuple))
-
-(defn -cleanup [this]
-  (.cleanup ^IBolt (. this state)))
-
-(defn -declareOutputFields [this declarer]
-  (.declareStream declarer ACKER-ACK-STREAM-ID true (Fields. ["id"]))
-  (.declareStream declarer ACKER-FAIL-STREAM-ID true (Fields. ["id"])))
-
-
diff --git a/src/clj/backtype/storm/daemon/common.clj b/src/clj/backtype/storm/daemon/common.clj
@@ -4,7 +4,6 @@
   (:import [backtype.storm.generated StormTopology])
   (:import [backtype.storm.utils Utils])
   (:require [backtype.storm.daemon.acker :as acker])
-  (:import [backtype.storm.daemon acker])
   (:require [backtype.storm.thrift :as thrift])
   )
 
@@ -109,7 +108,7 @@
                          [id ACKER-FAIL-STREAM-ID] ["id"]}
                         ))
         acker-bolt (thrift/mk-bolt-spec (merge spout-inputs bolt-inputs)
-                                        (backtype.storm.daemon.acker.)
+                                        (acker/mk-acker-bolt)
                                         :p (storm-conf TOPOLOGY-ACKERS))]
     (.put_to_bolts ret "__acker" acker-bolt)
     (dofor [[_ bolt] (.get_bolts ret)
diff --git a/src/clj/backtype/storm/testing.clj b/src/clj/backtype/storm/testing.clj
@@ -14,6 +14,7 @@
             SpoutTracker BoltTracker TrackerAggregator])
   (:require [backtype.storm [zookeeper :as zk]])
   (:require [backtype.storm.messaging.loader :as msg-loader])
+  (:require [backtype.storm.daemon.acker :as acker])
   (:use [clojure.contrib.def :only [defnk]])
   (:use [clojure.contrib.seq :only [find-first]])
   (:use [backtype.storm cluster util thrift config log]))
@@ -410,17 +411,16 @@
      }))
 
 (defmacro with-tracked-cluster [cluster-args & body]
-  ;;TODO: need an alternative approach to this
   `(with-var-roots [task/outbound-components (let [old# task/outbound-components]
                                                (fn [& args#]
                                                  (merge (apply old# args#)
                                                         {TrackerAggregator/TRACK_STREAM
                                                          {TRACKER-BOLT-ID (fn [& args#] 0)}}
                                                         )))
-                    task/mk-acker-bolt (let [old# task/mk-acker-bolt]
-                                         (fn [& args#]
-                                           (BoltTracker. (apply old# args#))
-                                           ))
+                    acker/mk-acker-bolt (let [old# acker/mk-acker-bolt]
+                                          (fn [& args#]
+                                            (BoltTracker. (apply old# args#))
+                                            ))
                     ]
      (with-local-cluster ~cluster-args
        ~@body
