                                                  (or new-opt old))
                                                new-verts g))
               old-messages messages
-              messages (.cache (g/aggregate-messages message-fn combiner g))]
+              messages (.cache (g/aggregate-messages wrapped-message-fn combiner g))]
           #_(println "Pregel iteration: " i "messages count: " (.count messages))
           (.count messages) ;; We must realise the messages
           (.unpersist old-messages false)

                    max-iterations Long/MAX_VALUE}}
              g]
  (let [dir (g/edge-directions direction)
        g (if initial-message
            (g/map-vertices
             (fn [vid attr] (vertex-fn vid attr initial-message)) g)
            g)]
    (loop [g g
           messages (g/aggregate-messages message-fn combiner g)
           i 0]
      (if (and (> (.count messages) 0)
               (< i max-iterations))

(ns glittering.pregel
  (:require  [glittering.core :as g]
             [glittering.destructuring :as g-de]
             [t6.from-scala.core :refer [$] :as $])
  (:import [org.apache.spark.graphx EdgeDirection]))

