             m-p2-next-acks (-> m-p2-acks
                                (m/flush-acks)
                                (m/receive-acks)
-                               (m/receive-acks)
-                               )]
-        (println "All acks seen"
-             (m/all-acks-seen? m-p1-acks)    
-                 )
+                               (m/receive-acks))]
         (is (m/all-acks-seen? m-p1-acks))
         (is (m/all-acks-seen? m-p2-acks))
         (is (m/all-acks-seen? m-p2-next-acks))))))
-
-
-;; Mutable version test - uses atom to replicate results in above test
-#_(deftest atom-messaging-test
-  ;; [:t2 :t1] [:t3 :t1]
-  (let [pg (component/start (am/atom-peer-group {}))
-        m-p1 (component/start (-> (am/atom-messenger)
-                                  (assoc :peer {:peer-group {:messaging-group pg}
-                                                :id :p1})))
-        m-p2 (component/start (-> (am/atom-messenger)
-                                  (assoc :peer {:peer-group {:messaging-group pg}
-                                                :id :p2})))
-        m-p3 (component/start (-> (am/atom-messenger)
-                                  (assoc :peer {:peer-group {:messaging-group pg}
-                                                :id :p3})))
-        t1-queue-p1 {:src-peer-id :p1 :dst-task-id :t1}
-        t1-queue-p2 {:src-peer-id :p2 :dst-task-id :t1}
-        t2-ack-queue {:src-peer-id :p3 :dst-task-id :t1}
-        _ (-> m-p1
-              (m/set-replica-version 1)
-              (m/add-publication t1-queue-p1)
-              (m/add-subscription t2-ack-queue))
-
-        _ (-> m-p2
-              (m/set-replica-version 1)
-              (m/add-publication t1-queue-p2)
-              (m/add-subscription t2-ack-queue))
-
-        m (-> m-p3
-              (m/set-replica-version 1)
-              (m/add-subscription t1-queue-p1)
-              (m/add-subscription t1-queue-p2)
-              (m/add-publication t2-ack-queue))
-
-        _ (-> m-p1 
-              (m/next-epoch)
-              (m/emit-barrier)
-              (m/send-messages [:m1 :m2] [t1-queue-p1])
-              (m/next-epoch)
-              (m/emit-barrier))
-
-        _ (-> m-p2
-              (m/next-epoch)
-              (m/emit-barrier)
-              (m/send-messages [:m3 :m4] [t1-queue-p2])
-              (m/next-epoch)
-              (m/emit-barrier)
-              (m/send-messages [:m5 :m6] [t1-queue-p2])) 
-        _ (process-barriers m-p3)
-        messages (map :message (mapcat (fn [_] (m/receive-messages m-p3)) (range 20)))]
-    (is (= [:m1 :m3 :m2 :m4] messages))
-    (is (m/all-barriers-seen? m))
-
-    ;; Because we've seen all the barriers we can call next epoch
-    ;; And continue reading the messages afterwards
-    (let [mnext (-> m-p3
-                    (m/ack-barrier)
-                    (m/next-epoch))
-          messages2 (map :message (mapcat (fn [_] (m/receive-messages m-p3)) (range 20)))]
-      (is (= [:m5 :m6] messages2))
-      (is (not (m/all-barriers-seen? mnext)))
-      (let [m-p1-acks (m/receive-acks m-p1)
-            m-p2-acks (m/receive-acks m-p2)
-            m-p2-next-acks (m/receive-acks m-p2)]
-        (is (not (empty? m-p1-acks)))
-        (is (not (empty? m-p2-acks)))
-        (is (empty? m-p2-next-acks))
-        ;; Late joiner :p4 on same queues at p3 should not obtain any messages
-        (let [m-p4 (component/start (-> (am/atom-messenger)
-                                        (assoc :peer {:peer-group {:messaging-group pg}
-                                                      :id :p4})))
-              m (-> m-p4
-                    (m/set-replica-version 1)
-                    (m/add-subscription t1-queue-p1)
-                    (m/add-subscription t1-queue-p2)
-                    (m/add-publication t2-ack-queue))]
-
-          (is (empty? (remove nil? (mapcat (fn [_] (m/receive-messages m-p4)) (range 20))))))))))

                     (m/emit-barrier)
                     (switch-peer :p4)
                     (m/receive-messages)
                     (m/receive-messages))]
        (is (m/all-barriers-seen? m-p4))

        ;(is (m/all-barriers-seen? (m/receive-messages (switch-peer (last mss) :p4))))
        )

      (let [m-p1-acks (-> (last mss)
                          (switch-peer :p1)
                          (m/receive-acks)
                          (m/receive-acks)
                          (m/receive-acks)
                          
                          )
            m-p2-acks (-> m-p1-acks 
                          (switch-peer :p2)


(defn switch-peer [messenger peer]
  (assoc messenger :peer-id peer))

(defn process-barriers [messenger] 
  (if (m/all-barriers-seen? messenger)
