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

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

(defn get-in-fn
  [pipeline]

  ([^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 out-counter out-drop-counter))
                      (catch InterruptedException e
                        (if @running
                          (throw e)))))
          thread-id (str "ProcessingElement_" id)
          proc-loop (if (not (nil? id))
                      (ProcessingLoop.
                        thread-id
                        proc-fn)
                      (ProcessingLoop.
                        proc-fn))]
