         conf {TOPOLOGY-WORKERS 6
               TOPOLOGY-MAX-SPOUT-PENDING 20
               TOPOLOGY-MESSAGE-TIMEOUT-SECS 60
-              TOPOLOGY-STATS-SAMPLE-RATE 1.00}]
+              TOPOLOGY-STATS-SAMPLE-RATE 1.00}
+        spout (doto (mk-fixed-batch-spout 3)
+                (.setCycle true))]
     (StormSubmitter/submitTopology
      name
      conf
      (.build (build-topology
+              spout
               nil)))))
 
 (defn -main

      (doseq [word words]
        (t/emit-fn coll word)))))

(defn build-topology [drpc]
  (let [spout (doto (mk-fixed-batch-spout 3)
                (.setCycle true))
        word-state-factory (MemoryMapState$Factory.)
        trident-topology (TridentTopology.)
        word-counts (-> (t/new-stream trident-topology "word-counts" spout)
                        (t/parallelism-hint 16)
                        (t/each ["sentence"]
                                 split-args
                                 ["word"])
                        (t/project ["word"])
                        (t/group-by ["word"])
                        (t/persistent-aggregate word-state-factory
                                                 ["word"]
                                                 count-words
                                                 ["count"])
                        (t/parallelism-hint 16))
        count-query (-> (t/drpc-stream trident-topology "words" drpc)
                        (t/each ["args"]
                                 split-args
                                 ["word"])
                        (t/project ["word"])
                        (t/group-by ["word"])
                        (t/state-query word-counts
                                        ["word"]
                                        (MapGet.)
                                        ["count"])
                        (t/each ["count"] (FilterNull.))
                        (t/aggregate ["count"]
                                      sum
                                      ["sum"])
                        (t/debug))]
    trident-topology))

(defn run-local! []
  (let [cluster (LocalCluster.)
        local-drpc (LocalDRPC.)]
    (.submitTopology cluster "wordcounter"
                     {}
                     (.build
                      (build-topology local-drpc)))
    (Thread/sleep 10000)
    (.execute local-drpc "words" "cat dog the man jumped")
    (.shutdown cluster)

(ns marceline.storm.testing.trident-wordcount
  (:import storm.trident.TridentTopology
           [storm.trident.operation.builtin
            MapGet
            FilterNull]
           storm.trident.state.StateType
           [storm.trident.testing MemoryMapState$Factory FixedBatchSpout]
           [backtype.storm LocalDRPC LocalCluster StormSubmitter])
  (:require [marceline.storm.trident :as t]
            [clojure.string :as string :only [join split]])
  (:use [backtype.storm clojure config])
