   (or (and (instance? Duration time) time)
       (.toStandardDuration time)))
 
+(def accumulation-mode-enum
+  {:accumulate #(.accumulatingFiredPanes %)
+   :discard #(.discardingFiredPanes %)})
+
+(def window-schema
+  (merge
+   named-schema
+   {:trigger {:docstr "Adds a Trigger to the Window."
+              :action (fn [transform ^Trigger t] (.triggering transform t))}
+    :with-allowed-lateness {:docstr "Allow late data. Mandatory for custom trigger"
+                            :action (fn [transform d] (.withAllowedLateness (->duration d)))}
+    :accumulate-mode {:docstr "Accumulate mode when a Trigger is fired (accumulate or discard)"
+                      :action (fn [transform acc] ((get accumulation-mode-enum acc) transform))}}))
+
 (defn sliding-windows
   {:doc (with-opts-docstr
           "Apply a sliding window input PCollection (useful for unbounded PCollections).

           [java.util UUID]
           [org.joda.time DateTimeUtils DateTimeZone]
           [org.joda.time.format DateTimeFormat DateTimeFormatter]
           [com.google.cloud.dataflow.sdk.transforms.windowing Window FixedWindows SlidingWindows Sessions]
           [org.joda.time Duration]))

(def required-ns (atom #{}))

  (ds/map (fn [e] (ds/with-timestamp (clj-time.core/now) (* 2 e)) pcoll))
  ```"
  [timestamp result]
  (->TimeStamped timestamp result))

(defn output-value!
  [^DoFn$ProcessContext context entity bindings]
