               ; because non-immediate reactions are buffered til join
               (let [replica (:replica state)
                     peerless-queues (->> (:entries state)
-                                         (filter (fn [[queue-id queue]] 
+                                         (filter (fn [[queue-id {:keys [predicate queue]}]] 
                                                    (or (peerless-entry? (first queue))
-                                                       (allow-queue? replica queue))))
+                                                       ((or predicate (constantly true)) 
+                                                        replica 
+                                                        (first queue)))))
                                          (map key))
                     joined-peers (set (:peers replica))
                     selectable-peers (->> (:entries state)
-                                          (filter (fn [[peer [entry]]]
-                                                    (or (:immediate? entry)
-                                                        (contains? joined-peers peer))))
+                                          (filter (fn [[peer {:keys [queue]}]]
+                                                    (let [entry (first queue)] 
+                                                      (or (:immediate? entry)
+                                                          (contains? joined-peers peer)))))
                                           (map key)
                                           set)
                     selectable-queues (into selectable-peers peerless-queues)]

     :entries updated-entries
     :log (conj log [message diff])
     :peer-choices (conj peer-choices next-peer)}))

;; This should be a predicate that can be setup on the queue
;; to decide when to allow the queue to progress - makes things easier
;; so you can start backpressure events and have known results
(defn allow-queue? [{:keys [peer-state] :as replica} 
                    [{:keys [args] :as entry} & es]] 
  (or (and (= :backpressure-on (:fn entry))
           (= :active (get peer-state (:peer args))))
      (= :backpressure-off (:fn entry))))

(defn queue-select-gen 
  "Generator to look into all of the peer's write queues


(defn generate-join-queues [peer-ids]
  (zipmap peer-ids
          (map (fn [peer] 
                 {:queue [(build-join-entry peer)]}) 
               peer-ids))) 

(defn generate-peer-ids [n]
  (map #(keyword (str "p" %))
