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

  (:require [dsbdp.data-processing-dsl :refer :all])
  (:import
    (dsbdp LocalTransferContainer ProcessingLoop)
    (java.util.concurrent ArrayBlockingQueue BlockingQueue LinkedTransferQueue)))







(defmacro create-queue
  []
  `(LinkedTransferQueue.))
;  `(ArrayBlockingQueue. *queue-size*))

(defn enqueue
  [^TransferQueue queue data]
;  [^BlockingQueue queue data]
;  (.offer queue data))
  (.tryTransfer queue data))



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