     ret))
 
 (def defaults
-  ["org.apache.hadoop.io.serializer.WritableSerialization"
-   "cascading.tuple.hadoop.BytesSerialization"
-   "cascading.tuple.hadoop.TupleSerialization"])
+  {"io.serializations"
+   (s/join "," ["org.apache.hadoop.io.serializer.WritableSerialization"
+                "cascading.tuple.hadoop.BytesSerialization"
+                "cascading.tuple.hadoop.TupleSerialization"])})
+
+(def mk-props
+  (partial merge defaults))
+
+(defn job-conf
+  "Returns a JobConf instance, optionally augmented by the supplied
+   property map."
+  ([] (job-conf {}))
+  ([prop-map]
+     (let [conf (JobConf.)]
+       (doseq [[k v] (mk-props prop-map)]
+         (.set conf k v))
+       conf)))
+
+(defn flow-connector
+  "Returns an instance of FlowConnection, optionally augmented by the
+   supplied property map."
+  ([] (flow-connector {}))
+  ([prop-map]
+     (FlowConnector. (mk-props prop-map))))
 
 (defn conj-serialization!
   "Appends the supplied serialization to the supplied JobConf
   object. Returns the modified JobConf object."
   [conf serialization]
-  (.set conf
-        "io.serializations"
-        (join "," (conj defaults serialization))))
-
-(def default-props)
-
-(defn jobconf
-  "Returns a JobConf instance, optionally augmented by the supplied
-   property map.."
-  ([] (jobconf {}))
-  ([prop-map]
-     (let [conf (JobConf.)]
-       (doseq [[k v] prop-map]
-         (.set conf k v))
-       conf)))
+  (let [old-val (.get conf "io.serializations")
+        new-val (str old-val "," serialization)]
+    (.set conf "io.serializations" new-val)))
 
 (defn populate!
   "Accepts a key-value tap, a sequence of key-value pairs (and an

  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 (jobconf 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] :or {props {}}}]
  (doto (.connect (FlowConnector. props) source sink pipe)
    (.complete)))

(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

  (test/with-fs-tmp [_ tmp]
    (-> (kv-tap tmp)
        (populate! pairs)
        (hfs->elephant! elephant-sink))))

(defn read-etap-with-flow [tap]
  (test/with-fs-tmp [fs tmp-path]
    (let [sink (kv-tap tmp-path)]
      (elephant->hfs! source sink)
      (tuple-seq sink))))

(defn just-tuples [actual]
  (chatty-checker [expected]
                  (= (set (tuple-seq actual))
                     (set expected))))

(tabular
 (fact "connect-test"
   (test/with-fs-tmp [_ src-tmp sink-tmp]     
     (let [sink (ElephantBaseTap. sink-tmp
                                  (kv-spec 4)
                                  (mk-options))]
       (fill-domain sink ?xs))))
 ?xs
 [[1 2] [3 4]]
 [["key" "val"] ["ham" "burger"]])

;; TODO: Invalid. Doesn't belong in this project; this makes far too
;; many assumptions about a thrift interface, etc. All we're concerned
