     (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

          status {}
          new-subscriber (sub/add-assembler 
                          (Subscriber. peer-id ticket-counters peer-config dst-task-id
                                       slot-id site batch-size read-bytes errors error bs channel
                                       conn sub lost-sessions sources short-id-status-pub 
                                       status-pubs nil nil nil status nil))]
      (info "Created subscriber" (sub/info new-subscriber))

    (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 error-counter error bs nil nil nil nil nil nil nil 
                 nil nil nil nil nil)) 
  (add-assembler [this]
    (set! assembler (ControlledFragmentAssembler. this))
