     (spout
       (nextTuple []
         (log/info "Polling :created >" @last-timestamp)
-        (let [new-votes (db/poll @last-timestamp)
+        (let [new-votes (poll @last-timestamp)
               grouped-votes (group-by :zid new-votes)]
           ; For each chunk of votes, for each conversation, send to the appropriate spout
           (doseq [[zid rxns] grouped-votes]

            [polismath.conv-man :as cm]
            [polismath.db :as db]
            [clojure.core.matrix :as ccm]
            [polismath.env :as env]
            [clojure.string :as string]
            [clojure.newtools.cli :refer [parse-opts]]

(ccm/set-current-implementation :vectorz)
(ccm/matrix [[1 2 3] [4 5 6]])

(let [at-a-time 100
      interval  1000
      reaction-gen
        (atom
          (sim/make-vote-gen 
            {:n-convs 3
             :vote-rate interval
             :person-count-start 4
             :person-count-growth 5
             :comment-count-start 3
             :comment-count-growth 3}))]
  (defn sim-poll! [& {:keys [last-timestamp] :as opts}]
    (let [results (take at-a-time @reaction-gen)]
      (swap! reaction-gen (partial drop at-a-time))
      (map (pc/fn-> (assoc :created (- last-timestamp (rand 500)))) results)))

  (defn poll [last-timestamp]
    (let [db-results (db/poll last-timestamp)
          last-timestamp (reduce max last-timestamp (map :created db-results))
          sim-results (sim-poll! :last-timestamp last-timestamp)]
      (concat db-results sim-results))))


(defspout sim-reaction-spout ["zid" "last-timestamp" "reactions"] {:prepare true}
  [conf context collector]
  (let [at-a-time 100
        interval  1000
        reaction-gen
          (atom
            (sim/make-vote-gen 
              {:n-convs 3
               :vote-rate interval
               :person-count-start 4
               :person-count-growth 5
               :comment-count-start 3
               :comment-count-growth 3}))]
    (spout
      (nextTuple []
        (log/info "Polling simutated reaction spout")
        (let [rxn-batch (take at-a-time @reaction-gen)
              split-rxns (group-by :zid rxn-batch)
              last-timestamp (System/currentTimeMillis)]
