            [com.google.cloud.dataflow.sdk.transforms
             DoFn DoFn$Context ParDo DoFnTester Create PTransform
             SerializableFunction WithKeys GroupByKey]
-           [com.google.cloud.dataflow.sdk.values PCollection TupleTag]
+           [com.google.cloud.dataflow.sdk.values PCollection TupleTag PBegin]
            [com.google.cloud.dataflow.sdk.coders ByteArrayCoder StringUtf8Coder CustomCoder Coder$Context]
            [com.google.cloud.dataflow.sdk.transforms.join KeyedPCollectionTuple CoGroupByKey])
   (:gen-class))

                   (TextIO$Write/to to))))))
  ([to pcoll] (write-text-file to {} pcoll)))

(defn- write-edn-file-transform
  [to options]
  (let [safe-opts (dissoc options :name)]
    (proxy [PTransform] []
      (apply [^PCollection pcoll]
        (->> pcoll
             (to-edn)
             (write-text-file to options))))))

(defn write-edn-file

(defn -main
  [& args]
  (let [p (make-pipeline args)
        proc (->> p
                  (read-edn-file "sample.edn" {:name :input-edn})
                  (dfilter (fn [{:keys [last-state] :as all}] (println all) (= last-state :sent)))
                  (write-edn-file "tessst.edn" {:name :output-edn}))
        ;; first-in (->> p
        ;;               (generate-input [{:key :a :foo 12} {:key :b :foo 5} {:key :a :foo 42} {:key :c :val "never"}] {:name "gengen1"}))
        ;; second-in (->> p
        ;;                (generate-input [{:key :a :val 10} {:key :b :val 5} {:key :a :val 42}] {:name "gengen"}))
        ;; final (->> (cogroup-by {:name :join} [[first-in :key] [second-in :key :required]] (fn [[k res]] (apply merge res)))
        ;;            (write-edn-file "tessst.edn" {:name :output-edn}))
        ;; first-in (->> p
        ;;               (generate-input [{:key :a :foo 12} {:key :b :foo 5} {:key :a :foo 42}] {:name "gengen1"})
        ;;               (dgroup-by :key {:name :group-by-fi}))
