 (def to-edn (partial (map-op identity (StringUtf8Coder/of)) to-edn*))
 
 (defn to-edn
-  [^PCollection pcoll]
-  (-> pcoll
-      (.apply (ParDo/of (dofn to-edn*)))
-      (.setCoder (StringUtf8Coder/of))))
+  ([opts ^PCollection pcoll]
+   (with-opts base-schema opts
+     (-> pcoll
+         (.apply (ParDo/of (dofn to-edn*)))
+         (.setCoder (StringUtf8Coder/of)))))
+  ([pcoll] (to-edn {} pcoll)))
 
 (defn make-pipeline
   [str-args]

(def dfilter (map-op filter-fn))

(defn generate-input
  [coll p]
  (-> p
      (.apply (Create/of (seq coll)))
      (.setCoder (make-transit-coder))))

(defn- to-edn*
  [^DoFn$Context c]

      (let [rdr (transit/reader in :msgpack)]
        (transit/read rdr)))))

(defmacro with-opts
  [schema opts & body]
  `(let [pcoll# (do ~@body)]
     (reduce
      (fn [pc# [k# f#]]
        (if-let [v# (get ~opts k#)]
          (f# pc# v#)
          pc#))
      pcoll# ~schema)))

(def base-schema
  {:named (fn [pcoll n] (.setName pcoll n))
   :coder (fn [pcoll coder] (.setCoder pcoll coder))})

(defn map-op
  ([transform coder]
   (fn
     [f opts ^PCollection pcoll]
     (with-opts base-schema opts
       (-> pcoll
           (.apply (ParDo/of (dofn (transform f))))
           (.setCoder coder)))))
  ([transform]
   (map-op transform (make-transit-coder))))

