           (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 start-subscriber! [shutdown bind-addr port stream-id virtual-peers decompress-f idle-strategy]
+  (future 
+   (loop []
+     (let [failed (atom false)
+           no-op-error-handler (reify ErrorHandler
+                                 (onError [this x] 
+                                   (reset! failed true)))
+           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)]
+       (try (.accept ^Consumer (consumer handler idle-strategy 10 shutdown failed) subscription)
+            (catch Throwable e 
+              (reset! failed true)
+              (warn "Subscriber failed. Restarting." e))
+            (finally
+             (.close subscription)
+             (.close conn))))
+     (when-not @shutdown
+       (info "Subscriber failed. Restarting.")
+       (recur)))
+   (info "Shutting down subscriber.")))
 
 (defn get-threading-model
   [media-driver]

          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)
          subscribers (mapv (fn [stream-id]
                              (start-subscriber! bind-addr port stream-id virtual-peers decompress-f receive-idle-strategy))
                            (range subscriber-count))]
      (when embedded-driver?
        (.addShutdownHook (Runtime/getRuntime)

  (when peer-task-id 
    (swap! virtual-peers pm/dissoc peer-task-id)))


(defn fragment-data-handler [f]
  (FragmentAssembler.
