              :messaging-group messaging-group
              :short-ids short-ids
              :send-idle-strategy send-idle-strategy
+             :publications publications
              :compress-f compress-f)))
 
-  (stop [{:keys [short-ids] :as component}]
+  (stop [{:keys [short-ids publications] :as component}]
     (taoensso.timbre/info "Stopping Aeron Messenger")
+    (run! #(.close %) (vals @publications))
     (assoc component
            :messaging-group nil
            :send-idle-strategy nil
+           :publications nil
            :short-ids nil
            :compress-f nil)))
 

    (taoensso.timbre/info "Starting Aeron Messenger")
    (let [config (:config peer-group)
          messaging-group (:messaging-group peer-group)
          publications (:publications messaging-group)
          send-idle-strategy (:send-idle-strategy messaging-group)
          compress-f (:compress-f (:messaging-group peer-group))
          short-ids (atom {})]

            fh (fragment-data-handler (partial handle-message result-state (:onyx.core/task-id event)
                                               (:upstream-task next-subscription)
                                               (:onyx.core/task event)))
            n-fragments (.controlledPoll ^Subscription subscription ^ControlledFragmentHandler fh fragment-limit)]
        @result-state)
      [])))

(defn get-publication [publications channel stream-id]
  (if-let [pub (get @publications [channel stream-id])]
    pub
    (let [error-handler (reify ErrorHandler
                          (onError [this x] 
                            (taoensso.timbre/warn "Aeron messaging publication error:" x)))
          ctx (-> (Aeron$Context.)
                  (.errorHandler error-handler))
          conn (Aeron/connect ctx)
          pub (.addPublication conn channel stream-id)]
      (swap! publications assoc [channel stream-id] pub)
      pub)))

(defn write [^Publication pub ^UnsafeBuffer buf]
  ;; Needs an escape mechanism so it can break if a peer is shutdown
  ;; Needs an idle mechanism to prevent cpu burn
  (while (neg? (.offer pub buf 0 (.capacity buf)))
    (info "Offering")))

(defmethod extensions/send-messages AeronMessenger
  [{:keys [publications]} {:keys [channel stream-id] :as conn-spec} batch]
  (let [pub (get-publication publications channel stream-id)
        buf ^UnsafeBuffer (UnsafeBuffer. (messaging-compress batch))]
    (write pub buf)))

(defmethod extensions/send-barrier AeronMessenger
  [{:keys [publications]} {:keys [channel stream-id] :as conn-spec} barrier]
  (let [pub (get-publication publications channel stream-id)
        buf ^UnsafeBuffer (UnsafeBuffer. (messaging-compress barrier))]
    (write pub buf)))

(defmethod extensions/internal-complete-segment AeronMessenger
  [{:keys [publications]} completion-message {:keys [channel stream-id] :as conn-spec}]
  (let [pub (get-publication publications channel stream-id)
        buf ^UnsafeBuffer (UnsafeBuffer. (messaging-compress completion-message))]
    (write pub buf)))
