 (def flow-conditions
   [{:flow/from :in
     :flow/to [:inc]
-    :flow/predicate :onyx.peer.flow-test/segment-even?
+    :flow/predicate [:and
+                     :onyx.peer.flow-test/segment-even?
+                     :onyx.peer.flow-test/gt?]
     :flow/doc "Emits segments to :inc if the segment's :n key is an even number."}])
 
 (defn segment-even? [event {:keys [n]}]
   (even? n))
+
+(defn gt? [event {:keys [n]}]
+  (> n 10))
 
 (def v-peers (onyx.api/start-peers! 1 peer-config))
 


(def env (onyx.api/start-env env-config))

(def n-messages 15000)

(def batch-size 1320)

(def echo 1000)


(doseq [v-peer v-peers]
  (onyx.api/shutdown-peer v-peer))

(let [expected (set (map (fn [x] {:n (inc x)}) (filter even? (range 11 n-messages))))]
  (fact (set (butlast results)) => expected)
  (fact (last results) => :done))

