         (println "RUNNING BOLT")
         (swap! data update-data conv-id reaction)
         (emit-bolt! collector
-                    [conv-id (:matrix (get @data conv-id))]
+                    [conv-id (ic.core/matrix (:matrix (get @data conv-id)))]
                     :anchor tuple))))))
+
+
+(defbolt ptpt-pca ["conv-id" "pcs" "proj"] {:prepare true}
+  [conf context collector]
+  ; If nothing exists yet, creating an empty matrix
+  (let [data (atom {})
+        n-comps 2]
+    (bolt (execute [tuple]
+      (let [[conv-id rating-matrix] (.getValues tuple)]
+        (swap! data (data-updater #(powerit-pca rating-matrix n-comps :start-vectors %)) conv-id)
+        (let [pcs (get @data conv-id)
+              proj (pca-project rating-matrix pcs)]
+          (emit-bolt! collector [conv-id pcs proj])))))))
+    ;{:pca pca :components components :pc1 pc1 :pc2 pc2 :x1 x1 :x2 x2})))))
+
+
+;(defbolt clusters ["clusters"] {:prepare true}
+  ;[conf context collector]
+  ;(let [
 
 
 (defn mk-topology []


(defspout reaction-spout ["conv-id" "reaction"] {:prepare true}
  [conf context collector]
(defn data-updater [update-fn init-fn]
  (fn [data conv-id inputs]
    (let [value (or (get data conv-id) (init-fn))]
      (update-fn value inputs))))

  (let [n-convs 3
        start-n 3
        reaction-gen (atom (make-reaction-gen n-convs start-n))]

      (ack [id]))))


(defn data-updater [update-fn & [init-fn]]
  "This isn't hooked up yet, but I'm hoping we can use something like this for
  abstracting the common pattern of storing/fetching results by conv-id"
  (let [init-fn #(identity nil)]
    (fn [data conv-id & inputs]
      (let [value (or (get data conv-id) (init-fn))
            new-value (apply update-fn value inputs)]
        (assoc data conv-id new-value)))))

(def init-matrix (->RatingMatrix ["p1" "p2" "p3"] ["c1" "c2" "c3"] [[1 1 0] [0 1 -1] [0 -1 1]]))

(defbolt rating-matrix ["conv-id" "rating-matrix"] {:prepare true}
  [conf context collector]
  (let [data (atom {})
        ; This is a function which we can call as a transaction to update our data atom
        update-data (fn [data conv-id reaction]
                      (let [rating-matrix (or (get data conv-id) init-matrix)
                            rating-matrix (update-rating-matrix rating-matrix [reaction])]
                        (assoc data conv-id rating-matrix)))]
    (bolt (execute [tuple]
