 (ns elephantdb.cascading.integration-test
   (:use midje.sweet
-        elephantdb.test.common
-        [clojure.string :only (join)])
+        elephantdb.test.common)
   (:require [elephantdb.test.keyval :as t]
-            [hadoop-util.test :as test])
+            [hadoop-util.test :as test]
+            [clojure.string :as s])
   (:import [cascading.pipe Pipe]
            [cascading.tuple Fields Tuple]
            [cascading.flow FlowConnector]

                 [(barr 2) (barr 2 2)]
                 [(barr 3) (barr 3)]]]
      (fill-domain sink data)
      (check-results tmp data)
      (fill-domain sink data2)
      (check-results tmp data3))))

(fact "test-source"
  (test/with-fs-tmp [fs tmp]

  and sinks all key-value pairs into the tap. Returns the original tap
  instance.."
  [kv-tap kv-pairs & {:keys [props]}]
  (with-open [collector (.openForWrite kv-tap (job-conf props))]
    (doseq [[k v] kv-pairs]
      (.add collector (Tuple. (into-array Object [k v])))))
  kv-tap)

(defn connect!
  "Connect the supplied source and sink with the supplied pipe."
  [pipe source sink & {:keys [props]}]
  (doto (.connect (flow-connector props) source sink pipe)
    (.complete)))

(defn tuple-seq
  "Returns all tuples in the supplied cascading tap as a Clojure
  sequence."
  [sink]
  (with-open [it (.openForRead sink (job-conf))]
    (doall (for [wrapper (iterator-seq it)]
             (into [] (.getTuple wrapper))))))

;; ## Transfer Functions

(defn elephant->hfs!
  "Transfers all tuples from the supplied elephant-tap into the
  supplied cascading `sink`."
  [elephant-source sink]
  (connect! (Pipe. "pipe")
            elephant-source
            sink))

(defn hfs->elephant!
  "Transfers all tuples from the supplied cascading `source` to the
  supplied elephant-tap."
  [source elephant-sink]
  (connect! (-> (Pipe. "pipe")
                (KeyValTailAssembly. elephant-sink))
            source
            elephant-sink))

(defn fill-domain
  "Fills the supplied elephant-sink with the the supplied sequence of
