 (defn -main
   [& args]
   (let [p (make-pipeline args)
-        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}))
+        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}))

                   (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

(ns datasplash.core
  (:require [cognitect.transit :as transit]
            [clojure.edn :as edn])
  (:import [datasplash.fns ClojureDoFn ClojureSerializableFn]
           [java.io InputStream OutputStream]
           [java.util UUID]
