     (onAvailableImage [this image] 
       (info "Available network image" (.position image) (.sessionId image) :correlation-id (.correlationId image) sub-info))))
 
+(defn new-error-handler [error error-counter]
+  (reify ErrorHandler
+    (onError [this x] 
+      (reset! error x)
+      (.addAndGet error-counter 1))))
+
 (deftype Subscriber 
   [peer-id
    ticket-counters

    (when conn (.close conn))
    (run! status-pub/stop (vals status-pubs))
    (Subscriber. peer-id ticket-counters peer-config dst-task-id slot-id site
                 batch-size read-bytes errors error bs nil nil nil nil nil nil nil 
                 nil nil nil nil nil)) 
  (add-assembler [this]
    (set! assembler (ControlledFragmentAssembler. this))

   ^:unsynchronized-mutable batch]
  sub/Subscriber
  (start [this]
    (let [media-driver-dir (:onyx.messaging.aeron/media-driver-dir peer-config)
          sinfo [dst-task-id slot-id site]
          lost-sessions (atom #{})
          ctx (cond-> (.errorHandler (Aeron$Context.)
                                     (new-error-handler error error-counter))
                media-driver-dir (.aeronDirectoryName ^String media-driver-dir))
          conn (Aeron/connect ctx)
          channel (autil/channel peer-config)
