   (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 peer-link unsafe-buffer 0 len))]
+        offer-f (fn [] (.offer (:pub peer-link) unsafe-buffer 0 len))]
     (while (not (offer-f)))))
 
 (defmethod extensions/internal-complete-message AeronConnection

  (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]
  {})


          acker-subscriber (.addSubscription aeron channel acker-stream-id acker-handler)
          completion-subscriber (.addSubscription aeron channel completion-stream-id completion-handler)]

      (future (try (.accept (consumer 10) send-subscriber) (catch Exception e (fatal e))))
      (future (try (.accept (consumer 10) acker-subscriber) (catch Exception e (fatal e))))
      (future (try (.accept (consumer 10) completion-subscriber) (catch Exception e (fatal e))))
      
      (assoc component :driver driver :channel channel :send-stream-id send-stream-id
             :acker-stream-id acker-stream-id :completion-stream-id completion-stream-id
             :release-ch release-ch :context ctx :aeron aeron)))

  (stop [component]
    (taoensso.timbre/info "Stopping Aeron")

    (.close (:aeron component))
    (CloseHelper/quietClose (:driver component))

    (close! (:release-ch component))

    (assoc component :driver nil :channel nil :release-ch nil)))

(defn aeron [opts]
