 (ns pigpen.cascading
   (:import (cascading.tap.hadoop Hfs)
            (cascading.scheme.hadoop TextLine)
-           (cascading.pipe Pipe))
+           (cascading.pipe Pipe)
+           (cascading.flow.hadoop HadoopFlowConnector))
   (:require [pigpen.raw :as raw]))
 
 (defn- get-tap-fn [name]

  (if (contains? (:sources flowdef) (first ancestors))
    (let [pipe (Pipe. (str id))]
      (-> flowdef
          (update-in [:source-to-pipe] (partial merge {(first ancestors) pipe}))
          (update-in [:pipes] (partial merge {id pipe}))))
    (throw (Exception. "not implemented"))))


(defmethod command->flowdef :store
           [{:keys [id ancestors location storage opts]} flowdef]
  {:pre [id ancestors location storage]}
  (-> flowdef
      (update-in [:pipe-to-sink] (partial merge {(first ancestors) id}))
      (update-in [:sinks] (partial merge {id ((get-tap-fn (:func storage)) location)}))))

(defmethod command->flowdef :generate
           [{:keys [id ancestors projections opts]} flowdef]
