 
 (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)
+  [pipe source sink]
+  (doto (.connect (flow-connector) source sink pipe)
     (.complete)))
 
 (defn tuple-seq

  optional JobConf instance, supplied with the :conf keyword argument)
  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 job-conf
  "Returns a JobConf instance, optionally augmented by the supplied
   property map."
  [& [prop-map]]
  (let [conf (JobConf.)]
    (doseq [[k v] (mk-props (or prop-map *default-conf*))]
      (.set conf k v))
    conf))

(defn flow-connector
  "Returns an instance of FlowConnection, optionally augmented by the
   supplied property map."
  [& [prop-map]]
  (FlowConnector. (mk-props (or prop-map *default-conf*))))

(defn conj-serialization!
  "Appends the supplied serialization to the supplied configuration
  map. object. Returns the modified JobConf object."
  [conf-map serialization]
  (merge-with #(str %1 "," %2)
              conf-map
              {"io.serializations"  serialization}))

(let [conf {"io.serializations" "StringWrapper"}]
  (fact
    (conj-serialization! {} "CakeWrapper") => {"io.serializations" "CakeWrapper"}
    "Serializations should join up with commas."
    (conj-serialization! conf "CakeWrapper") => {"io.serializations"
                                                 "StringWrapper,CakeWrapper"}))

(defn elephant-tap
  "Returns an ElephantDB Tap tuned to the supplied path and
