           ;                      dest-pubs (m/task->publishers messenger dst-task-id)]
           ;                  (get dest-pubs (mod hsh (count dest-pubs))))
           ;                (rand-nth (m/task->publishers messenger dst-task-id))))
-          dst-task->publisher (m/task->publishers messenger)
-          get-pub-fn (fn [dst-task-id] (get dst-task->publisher dst-task-id))
           ;; TODO, avoid initial flattening preprocessing step
-          _ (run! (fn [{:keys [leaves] :as result}]
+          _ (run! (fn [result]
                     (run! (fn [seg]
-                            (add-segment segments seg event result get-pub-fn))
-                          leaves))
+                            (add-segment segments routes size seg event result))
+                          (:leaves result)))
                   (:tree results))
           _ (run! (fn [seg] 
-                    (add-segment segments seg event {:leaves [seg]} get-pub-fn))
+                    (add-segment segments routes size seg event {:leaves [seg]}))
                   triggered)]
       (run! pub/reset-segment-encoder! (m/publishers messenger))
-      (set! iterator (.iterator segments))
+      (set! queued-offers nil)
+      (set! full-pub nil)
       true))
 
   (write-batch [this event replica messenger]
-    ;; FIXME rename add failed to FULL PUB OR SOMETHING
-    (cond add-failed
-          ;; previous addition of segment to buffer resulted in an overflow
-          ;; offer the full buffer, and then add the serialized segment to the reset buffer
-          (if (neg? (pub/offer-segments! add-failed))
-            false
-            (do (pub/reset-segment-encoder! add-failed)
-                ;; publisher already has the last segment saved, lets encode it now
-                (when-not (pub/encode-segment! add-failed)
-                  (throw (ex-info "Aeron buffer size is not big enough to contain the segment.
-                                   Please increase the term buffer length via java property aeron.term.buffer.length"
-                                  {})))
-                (set! add-failed nil)
-                (p/write-batch this event replica messenger)))
+    (if (zero? (.get size))
+      true
+      (let [dst-task->publisher (m/task->publishers messenger)
+            get-pub-fn (fn [dst-task-id] (get dst-task->publisher dst-task-id))]
+        (cond full-pub
+              ;; previous addition of segment to buffer resulted in an overflow
+              ;; offer the full buffer, and then add the serialized segment to the reset buffer
+              (if (neg? (pub/offer-segments! full-pub))
+                false
+                (do (pub/reset-segment-encoder! full-pub)
+                    ;; publisher already has the last segment saved, lets encode it now
+                    (when-not (pub/encode-segment! full-pub)
+                      (throw (ex-info "Aeron buffer size is not big enough to contain the segment.
+                                       Please increase the term buffer length via java property aeron.term.buffer.length"
+                                      {})))
+                    (set! full-pub nil)
+                    (p/write-batch this event replica messenger)))
 
-          (.hasNext iterator)
-          ;; add segment to corresponding buffer
-          (loop [] 
-            (if (.hasNext iterator)
-              ;; SLOW GETTING OUT OF THIS LIST
-              (let [[segment publisher] (.next iterator)]
-                (if (pub/encode-segment! publisher segment)
-                  (recur)
-                  (do (set! add-failed publisher)
-                      (p/write-batch this event replica messenger))))
-              (p/write-batch this event replica messenger)))
+              (< (.get idx) (.get size))
+              (loop [i (.get idx)] 
+                (if (< i (.get size))
+                  ;; add segment to corresponding buffer in publication
+                  (let [segment (aget segments i)
+                        route (aget routes i)
+                        publisher ((get-pub-fn route) segment)]
+                    (if (pub/encode-segment! publisher segment)
+                      (recur (.incrementAndGet idx))
+                      (do (set! full-pub publisher)
+                          (p/write-batch this event replica messenger))))
+                  (do
+                   (set! queued-offers (m/publishers messenger))
+                   (p/write-batch this event replica messenger))))
 
-          ;; iterator is empty, move to final phase where we send main buffers out
-          (empty? queued-offers)
-          (do (set! queued-offers (m/publishers messenger))
-              (p/write-batch this event replica messenger))
-
-          :else
-          (let [remaining (try-sends queued-offers)]
-            (set! queued-offers remaining)
-            (if (empty? remaining)
-              (do (.clear segments)
-                  true)
-              false)))))
+              :else
+              (let [remaining (try-sends queued-offers)]
+                (set! queued-offers remaining)
+                (if (empty? remaining)
+                  (do (run! (fn [i] 
+                              (aset segments i nil)
+                              (aset routes i nil))
+                            (range (.get size)))
+                      true)
+                  false)))))))
 
 (defn new-messenger-output [{:keys [onyx.core/task-map] :as event}]
   (let [bs (byte-array (max-message-length)) 
-        segments (java.util.ArrayList. 2000)]
-    (->MessengerOutput segments nil nil nil)))
+        segments (object-array 20000)
+        routes (make-array clojure.lang.Keyword 20000)]
+    (->MessengerOutput segments routes (AtomicInteger. 0) (AtomicInteger. 0) nil nil)))

  p/BarrierSynchronization
  (synced? [this _]
    true)
  (completed? [this] (empty? segments))

  p/Output
  (prepare-batch [this {:keys [onyx.core/results onyx.core/triggered task->group-by-fn] :as event} 
                  replica messenger]
    (.clear ^java.util.ArrayList segments)
    (let [;; generate this on each new replica / messenger
          ; get-pub-fn (fn [segment dst-task-id]
          ;              (if-let [group-fn (task->group-by-fn dst-task-id)]

        pubs
        (recur (rest pubs))))))

(deftype MessengerOutput [#^"[Ljava.lang.Object;" segments 
                          #^"[Lclojure.lang.Keyword;" routes
                          ^AtomicInteger idx
                          ^AtomicInteger size
                          ^:unsynchronized-mutable queued-offers
                          ^:unsynchronized-mutable full-pub]
  p/Plugin
  (start [this event] this)
  (stop [this event] this)
