           virtual-peers (atom (pm/vpeer-manager))
           publication-pool (component/start (pub-pool/new-publication-pool opts send-idle-strategy))
           subscriber-count (arg-or-default :onyx.messaging.aeron/subscriber-count opts)
+          shutdown (atom false)
           subscribers (mapv (fn [stream-id]
-                              (start-subscriber! bind-addr port stream-id virtual-peers decompress-f receive-idle-strategy))
+                              (start-subscriber! shutdown bind-addr port stream-id virtual-peers decompress-f receive-idle-strategy))
                             (range subscriber-count))]
       (when embedded-driver?
         (.addShutdownHook (Runtime/getRuntime)

          (throw (ex-info "Could not deserialize incoming message from Aeron"
                          {:msg-type msg-type})))))

(defn start-subscriber! [bind-addr port stream-id virtual-peers decompress-f idle-strategy]
  (let [ctx (-> (Aeron$Context.)
                (.errorHandler no-op-error-handler))
        conn (Aeron/connect ctx)
        channel (aeron-channel bind-addr port)
        handler (fragment-data-handler
                  (fn [buffer offset length header]
                    (handle-message decompress-f virtual-peers buffer offset length header)))
        subscription (.addSubscription conn channel stream-id)
        subscriber-fut (future (try (.accept ^Consumer (consumer handler idle-strategy 10) subscription)
                                    (catch Throwable e (fatal e))))]
    {:conn conn :subscription subscription :subscriber-fut subscriber-fut}))

(defn get-threading-model
  [media-driver]

             :virtual-peers virtual-peers
             :compress-f compress-f
             :decompress-f decompress-f
             :shutdown shutdown
             :port port
             :send-idle-strategy send-idle-strategy
             :subscriber-count subscriber-count
             :subscribers subscribers)))

  (stop [{:keys [media-driver media-driver-context subscribers publication-pool shutdown] :as component}]
    (taoensso.timbre/info "Stopping Aeron Peer Group")
    (component/stop publication-pool)
    (reset! shutdown true)
     
    (when media-driver (.close ^MediaDriver media-driver))
    (when media-driver-context (.deleteAeronDirectory ^MediaDriver$Context media-driver-context))
