   dsbdp.local-data-processing-pipeline
   (:require [dsbdp.data-processing-dsl :refer :all])
   (:import
-    (dsbdp LocalTransferContainer ProcessingLoop)
+    (dsbdp Counter LocalTransferContainer ProcessingLoop)
     (java.util.concurrent ArrayBlockingQueue BlockingQueue LinkedBlockingQueue LinkedTransferQueue TransferQueue)))
 
 

  ([^BlockingQueue in-queue f id]
    (let [running (atom true)
          out-queue (create-queue)
          proc-fn (fn []
                    (try
                      (let [^LocalTransferContainer c (take-from-queue in-queue)
                            new-out (f (.getIn c) (.getOut c))]
                        (if (not (nil? new-out))
                          (.setOut c new-out))
                        (enqueue out-queue c))
                      (catch InterruptedException e
                        (if @running
                          (throw e)))))
          proc-loop (if (not (nil? id))
                      (ProcessingLoop.
                        (str "ProcessingElement_" id)
                        proc-fn)
                      (ProcessingLoop.
                        proc-fn))]

      {:interrupt (fn []
                    (reset! running false)
                    (.interrupt proc-loop))
       :out-queue out-queue
       :thread-id thread-id})))

(defn interrupt
  [proc-element]
