   [messenger event {:keys [channel stream-id]}]
   (let [ctx (.errorHandler (Aeron$Context.) no-op-error-handler)
         aeron (Aeron/connect ctx)]
-    (.addPublication aeron channel stream-id)))
+    {:ctx ctx
+     :conn aeron
+     :pub (.addPublication aeron channel stream-id)}))
 
 (defmethod extensions/receive-messages AeronConnection
   [messenger {:keys [onyx.core/task-map] :as event}]

  (let [compressed (compress {:id id})
        len (count compressed)
        unsafe-buffer (UnsafeBuffer. compressed)
        offer-f (fn [] (.offer peer-link unsafe-buffer 0 len))]
    (while (not (offer-f)))))

(defmethod extensions/close-peer-connection AeronConnection
  [messenger event peer-link]
  {})


  (let [compressed (compress {:id message-id :completion-id completion-id :ack-val ack-val})
        len (count compressed)
        unsafe-buffer (UnsafeBuffer. compressed)
        offer-f (fn [] (.offer (:pub peer-link) unsafe-buffer 0 len))]
    (while (not (offer-f)))))

(defmethod extensions/internal-complete-message AeronConnection
