 ;;; delay is rather long
 
 (def input
-  [{:id 1  :age 21 :event-time #inst "2015-09-13T03:00:00.829-00:00"}
-   {:id 2  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
-   {:id 3  :age 3  :event-time #inst "2015-09-13T03:05:00.829-00:00"}
-   {:id 4  :age 64 :event-time #inst "2015-09-13T03:06:00.829-00:00"}
-   {:id 5  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
-   {:id 6  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
-   {:id 7  :age 24 :event-time #inst "2015-09-13T03:09:00.829-00:00"}
-   {:id 8  :age 35 :event-time #inst "2015-09-13T03:15:00.829-00:00"}
-   {:id 9  :age 49 :event-time #inst "2015-09-13T03:25:00.829-00:00"}
-   {:id 10 :age 37 :event-time #inst "2015-09-13T03:45:00.829-00:00"}
-   {:id 11 :age 15 :event-time #inst "2015-09-13T03:03:00.829-00:00"}
-   {:id 12 :age 22 :event-time #inst "2015-09-13T03:56:00.829-00:00"}
-   {:id 13 :age 83 :event-time #inst "2015-09-13T03:59:00.829-00:00"}
-   {:id 14 :age 60 :event-time #inst "2015-09-13T03:32:00.829-00:00"}
-   {:id 15 :age 35 :event-time #inst "2015-09-13T03:16:00.829-00:00"}])
+  (reduce into [] 
+          (repeat 20
+                  [{:id 1  :age 21 :event-time #inst "2015-09-13T03:00:00.829-00:00"}
+                   {:id 2  :age 12 :event-time #inst "2015-09-13T03:04:00.829-00:00"}
+                   {:id 3  :age 3  :event-time #inst "2015-09-13T03:05:00.829-00:00"}
+                   {:id 4  :age 64 :event-time #inst "2015-09-13T03:06:00.829-00:00"}
+                   {:id 5  :age 53 :event-time #inst "2015-09-13T03:07:00.829-00:00"}
+                   {:id 6  :age 52 :event-time #inst "2015-09-13T03:08:00.829-00:00"}
+                   {:id 7  :age 24 :event-time #inst "2015-09-13T03:09:00.829-00:00"}
+                   {:id 8  :age 35 :event-time #inst "2015-09-13T03:15:00.829-00:00"}
+                   {:id 9  :age 49 :event-time #inst "2015-09-13T03:25:00.829-00:00"}
+                   {:id 10 :age 37 :event-time #inst "2015-09-13T03:45:00.829-00:00"}
+                   {:id 11 :age 15 :event-time #inst "2015-09-13T03:03:00.829-00:00"}
+                   {:id 12 :age 22 :event-time #inst "2015-09-13T03:56:00.829-00:00"}
+                   {:id 13 :age 83 :event-time #inst "2015-09-13T03:59:00.829-00:00"}
+                   {:id 14 :age 60 :event-time #inst "2015-09-13T03:32:00.829-00:00"}
+                   {:id 15 :age 35 :event-time #inst "2015-09-13T03:16:00.829-00:00"}])))
 
 (def expected-windows
   [[1442113200000 1442113499999 1] [1442113200000 1442113499999 2] [1442113500000 1442113799999 1] [1442113500000 1442113799999 2] [1442113500000 1442113799999 3] [1442114100000 1442114399999 1] [1442115000000 1442115299999 1] [1442116500000 1442116799999 1] [1442116500000 1442116799999 2] [1442113200000 1442113499999 3] [1442115900000 1442116199999 1] [1442114700000 1442114999999 1] [1442114100000 1442114399999 2] [1442113500000 1442113799999 4] [1442113500000 1442113799999 5]])


(defn update-atom! [event window-id lower-bound upper-bound state]
  (swap! test-state conj [lower-bound upper-bound state]))

(def in-chan (chan (inc (count input))))

(def out-chan (chan (sliding-buffer (inc (count input)))))

(defn inject-in-ch [event lifecycle]
  {:core.async/chan in-chan})

(defn inject-out-ch [event lifecycle]
  {:core.async/chan out-chan})

(def in-calls
  {:lifecycle/before-task-start inject-in-ch})

(def out-calls
  {:lifecycle/before-task-start inject-out-ch})

(defn restartable? [e] 
  true)

   (fn [event lifecycle]
     ; give the peer a bit of time to write the chunks out and ack the batches,
     ; since we want to ensure that the batches aren't re-read on restart
     ;{}
     (when (= (mod (swap! batch-num inc) 4) 0)
       (Thread/sleep 7000)
       (throw (ex-info "Restartable" {:restartable? true}))))})

(defrecord MonitoringStats
  [zookeeper-write-log-entry
   zookeeper-read-log-entry
   zookeeper-write-catalog
   zookeeper-write-workflow
   zookeeper-write-flow-conditions
   zookeeper-write-lifecycles
   zookeeper-write-task
   zookeeper-write-chunk
   zookeeper-write-job-scheduler
   zookeeper-write-messaging
   zookeeper-force-write-chunk
   zookeeper-write-origin
   zookeeper-read-catalog
   zookeeper-read-workflow
   zookeeper-read-flow-conditions
   zookeeper-read-lifecycles
   zookeeper-read-task
   zookeeper-read-chunk
   zookeeper-read-origin
   zookeeper-read-job-scheduler
   zookeeper-read-messaging
   zookeeper-gc-log-entry
   window-log-write-entry
   window-log-playback
   peer-ack-segments
   peer-retry-segment
   peer-try-complete-job
   peer-strip-sentinel
   peer-complete-message
   peer-gc-peer-link
   peer-backpressure-on
   peer-backpressure-off
   peer-prepare-join
   peer-notify-join
   peer-accept-join])


(defn identity-mod-id [segment]
  [segment (update segment :id * 10000)])

(deftest fault-tolerance-fixed-windows-segment-trigger

  (def in-chan (chan (inc (count input))))

  (def out-chan (chan (sliding-buffer (inc (count input)))))

  (defn inject-in-ch [event lifecycle]
    {:core.async/chan in-chan})

  (defn inject-out-ch [event lifecycle]
    {:core.async/chan out-chan})

  (def in-calls
    {:lifecycle/before-task-start inject-in-ch})

  (def out-calls
    {:lifecycle/before-task-start inject-out-ch})


  (let [id (java.util.UUID/randomUUID)
        config (load-config)
        env-config (assoc (:env-config config) :onyx/id id)
        peer-config (assoc (:peer-config config) :onyx/id id)
        batch-size 10
        workflow
        [[:in :identity] [:identity :out]]

