             [midje.sweet :refer :all]
             [onyx.peer.task-lifecycle-extensions :as l-ext]
             [onyx.plugin.core-async :refer [take-segments!]]
+            [onyx.extensions :as extensions]
             [onyx.api]))
 
 (def onyx-id (java.util.UUID/randomUUID))

    :onyx/type :output
    :onyx/medium :core.async
    :onyx/batch-size 20
    :onyx/max-peers 1
    :onyx/doc "Writes segments to a core.async channel"}])

(def n-messages 1000)

          entry (extensions/read-log-entry (:log env) position)
          new-replica (extensions/apply-log-entry entry replica)]
      (if-not (and (= (count (get (get (:allocations new-replica) j1) task-a)) 0)
                   (= (count (get (get (:allocations new-replica) j1) task-b)) 0)
                   (= (count (get (get (:allocations new-replica) j1) task-c)) 0)
                   (= (count (get (get (:allocations new-replica) j2) task-d)) 12)
                   (= (count (get (get (:allocations new-replica) j2) task-e)) 12)
                   (= (count (get (get (:allocations new-replica) j2) task-f)) 12))
        (recur new-replica)
        new-replica))))

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