 
 (defmethod job-fn :source
   [node conf jname]
-  (fn [& args] (-> node :source first)))
+  (constantly (:source node)))
 
 (defmethod job-fn :default
   [node conf jname]

  "Configure a job according to the state of a job graph node."
  [^Job job node]
  (doto job
    (source-config (:source node))
    (subnodes-config node (:subnodes node))
    (pgt/mapper! node)
    (pgt/combiner! node)


(defn source
  "Return a fresh `:source`-stage job graph node consuming from the
provided `dseq`."
  [dseq]
  {:stage :source,
   :source-id (gen-id),
   :source (dseq/dseq dseq),
   :config [],
   })

(defn config
  "Add arbitrary configuration steps to current job graph `node`."
