       (extensions/touch-place sync completion-node))
     (assoc event :completed done?)))
 
-(defn open-session-loop [read-ch pipeline-data]
-  (loop []
-    (when-let [session (create-tx-session pipeline-data)]
-      (>!! read-ch (munge-open-session pipeline-data session))
-      (recur))))
+(defn open-session-loop [read-ch kill-ch pipeline-data]
+  (loop []
+    (when (first (alts!! [kill-ch] :default true))
+      (when-let [session (create-tx-session pipeline-data)]
+        (>!! read-ch (munge-open-session pipeline-data session))
+        (recur)))))
 
 (defn read-batch-loop [read-ch decompress-ch]
   (loop []

  (stop [component]
    (prn "Stopping Task Pipeline")

    (future-cancel (:open-session-loop component))

    (close! (:read-batch-ch component))
    (close! (:decompress-batch-ch component))
    (close! (:apply-fn-ch component))

        (fn [e & _] (.printStackTrace e)))

      (assoc component
        :open-session-kill-ch open-session-kill-ch
        :read-batch-ch read-batch-ch
        :decompress-batch-ch decompress-batch-ch
        :apply-fn-ch apply-fn-ch
