     ; Spouts:
     {"1" (spout-spec reaction-spout)}
     ; Bolts:
-    {"2" (bolt-spec {"1"  ["conv-id"]}
-                    rating-matrix)
-     "3" (bolt-spec {"2" ["conv-id"]}
-                    ptpt-pca)
-     }))
+    {"2" (bolt-spec
+           {"1"  ["conv-id"]}
+           conv-update)}))
 
 
 (defn run-local! []

(ns polismath.storm-spec
  (:import [backtype.storm StormSubmitter LocalCluster])
  (:require [incanter.core :as ic.core])
  (:use [backtype.storm clojure config]
        polismath.matrix-utils
        polismath.pca
        polismath.simulation)
  (:gen-class))


            new-value (apply update-fn value inputs)]
        (assoc data conv-id new-value)))))


(defbolt conv-update ["conv"] {:prepare true}
  [conf context collector]
  (let [convs (atom {})]
    (bolt (execute [tuple]
      (let [[conv-id reaction] (.getValues tuple)]
        (swap! convs
               (data-updater
                 #(conv-update % [reaction]) hash-map)
                 conv-id)
        (emit-bolt! collector
                    [(get @data conv-id)]
                    :anchor tuple)
        (ack! collector tuple))))))


(defn mk-topology []
