                     (interrupt pe))
                   (.interrupt out-proc-loop))
      :in-fn (fn [in]
-              (.put in-queue (LocalTransferContainer. in nil)))}))
+              (enqueue in-queue (LocalTransferContainer. in nil)))}))
 
 (defn get-in-fn
   [obj]




(defn create-local-processing-element
  [^BlockingQueue in-queue f]
  (let [running (atom true)
        out-queue (LinkedTransferQueue.)
        proc-loop (ProcessingLoop.
                    (fn []
                      (try

                              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))))))]
