   ([^BlockingQueue in-queue f id]
     (let [running (atom true)
           out-queue (create-queue)
+          out-counter (Counter.)
+          out-drop-counter (Counter.)
           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))
+                        (enqueue out-queue c out-counter out-drop-counter))
                       (catch InterruptedException e
                         (if @running
                           (throw e)))))
+          thread-id (str "ProcessingElement_" id)
           proc-loop (if (not (nil? id))
                       (ProcessingLoop.
-                        (str "ProcessingElement_" id)
+                        thread-id
                         proc-fn)
                       (ProcessingLoop.
                         proc-fn))]

                    (interrupt pe))
                  (.interrupt out-proc-loop))
     :in-fn (fn [in]
              (enqueue in-queue (LocalTransferContainer. in nil)))}))

(defn get-in-fn
  [pipeline]

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

(defn interrupt
  [proc-element]
