 
 (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

(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]
  (let [old-val (.get conf "io.serializations")
        new-val (str old-val "," serialization)]
    (.set conf "io.serializations" new-val)))

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

  `with-kv-tap` accepts a `:log-level` optional keyword argument that
  can be used to tune the output of all jobs run within the
  form. Valid log level values or `:fatal`, `:warn`, `:info`, `:debug`
  and `:off`.

  To change the configuration map used for the test, supply a map
  using the `:conf` keyword argument."
  [[sym shard-count & opts] & body]
  (let [opt-map   (apply hash-map opts)
        log-level (:log-level opt-map :off)
        conf      (:conf opt-map *default-conf*)]
    `(binding [*default-conf* (or ~conf)]
       (log/with-log-level ~log-level
         (test/with-fs-tmp [fs# tmp#]
           (let [~sym (elephant-tap tmp# ~shard-count ~@opts)]
             ~@body))))))

;; ## Tests

