 (defmethod extensions/internal-complete-message AeronConnection
   [messenger event id peer-link]
   (let [unsafe-buffer (protocol/build-completion-msg-buf id)
-        pub ^uk.co.real_logic.aeron.Publication (:aux-pub peer-link) 
+        pub ^uk.co.real_logic.aeron.Publication (get-peer-link-pub peer-link :aux-pub)
         offer-f (fn [] (.offer pub unsafe-buffer 0 protocol/completion-msg-length))]
     (while (not (offer-f))
       (.idle ^IdleStrategy (:send-idle-strategy messenger) 0))))

   :aeron/external-addr (:external-addr messenger)})

(def send-stream-id 1)
(def acker-stream-id 2)
(def completion-stream-id 3)
(def retry-stream-id 4)
(def aux-stream-id 5)

(defn aeron-channel [addr port]
  (format "udp://%s:%s" addr port))

(defmethod extensions/internal-retry-message AeronConnection
  [messenger event id peer-link]
  (let [unsafe-buffer (protocol/build-retry-msg-buf id)
        pub ^uk.co.real_logic.aeron.Publication (get-peer-link-pub peer-link :aux-pub)
        offer-f (fn [] (.offer pub unsafe-buffer 0 protocol/retry-msg-length))]
    (while (not (offer-f))
      (.idle ^IdleStrategy (:send-idle-strategy messenger) 0))))

(defmethod extensions/close-peer-connection AeronConnection
  [messenger event peer-link]
  (when-let [pub @(get-in peer-link [:send-pub :pub])] 
    (.close pub))
  (when-let [pub @(get-in peer-link [:send-pub :aux-pub])] 
    (.close pub))
  (.close (:conn peer-link)) 
  {})
