gistfile1.txt

(def *data* [2 4 8 16 32 64 128 256 512 1024 4096 2 4 8 16 32 64 128 256 512 1024 4096])

(defspout number-spout ["number"]
  [conf context collector]
  (spout
   (nextTuple []
              (Thread/sleep 100)
              (emit-spout! collector [(next *data*)]))
   (ack [id])))

(defbolt neg-bolt ["neg"] [tuple collector]
  (let [result (- (.getInteger tuple 0) )]
    (emit-bolt! collector [result] :anchor tuple :p 4)
    (ack! collector tuple)))

(defn mk-topology []
  (topology
   {1 (spout-spec number-spout)}
   {2 (bolt-spec {1 :shuffle} neg-bolt)}))

(defn run-local! []
  (let [cluster (LocalCluster.)]
    (.submitTopology cluster "test" {TOPOLOGY-DEBUG true} (mk-topology))
    (Thread/sleep 1000)
    (.shutdown cluster)))