 (require '[clj-time.core :as time])
 (ds/session-windows (time/minutes 10) pcoll)
 ```"
-          named-schema)
+          window-schema)
    :added "0.4.1"}
   ([gap options ^PCollection pcoll]
    (let [transform (-> (->duration gap)

(require '[clj-time.core :as time])
(ds/sliding-windows (time/minutes 30) (time/seconds 5) pcoll)
```"
          named-schema)
   :added "0.4.1"}
  ([width step options ^PCollection pcoll]
   (let [transform (-> (->duration width)

  (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).
