 
 (def ages (map #(select-keys % [:id :age]) people))
 
-(def name-chan (chan (inc names)))
+(def name-chan (chan (inc (count names))))
 
-(def age-chan (chan (inc ages)))
+(def age-chan (chan (inc (count ages))))
 
-(def out-chan (chan (sliding-buffer (inc n-messages))))
+(def out-chan (chan 10000 #_(sliding-buffer (inc n-messages))))
 
 (doseq [name names]
   (>!! name-chan name))


(def results (take-segments! out-chan))

(fact (into #{} (butlast results)) => (into #{} people))

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

  [_ event]
  {:onyx.core/params [(atom {})]})

(defmethod l-ext/inject-lifecycle-resources :names
  [_ _] {:core.async/chan name-chan})

(defmethod l-ext/inject-lifecycle-resources :ages
  [_ _] {:core.async/chan age-chan})

(defmethod l-ext/inject-lifecycle-resources :out
  [_ _] {:core.async/chan out-chan})

(def v-peers (onyx.api/start-peers 4 peer-group))

(onyx.api/submit-job
