                (stream-id job-id dst-task-id slot-id site)
                nil nil nil))
 
-(defn add-to-subscriptions [subscriptions sub-info]
-  (conj (or subscriptions []) sub-info))
-
-(defn remove-from-subscriptions 
-  [subscriptions sub-info]
-  {:post [(= (dec (count subscriptions)) (count %))]}
-  (let [to-remove (first (filter #(sub/equiv-meta % sub-info) subscriptions))] 
-    (assert to-remove)
-    (sub/stop to-remove)
-    (vec (remove #(sub/equiv-meta % sub-info) subscriptions))))
-
-(defn remove-from-publications [publications pub-info]
-  {:pre [(pos? (count publications))]
-   :post [(= (dec (count publications)) (count %))]}
-  (println "Count publications" (count publications))
-  (let [to-remove (first (filter #(pub/equiv-meta % pub-info) publications))] 
-    (assert to-remove)
-    (pub/stop to-remove)
-    (vec (remove #(pub/equiv-meta % pub-info) publications))))
-
 ;; TICKETS SHOULD USE session id (unique publication) and position
 ;; Lookup task, then session id, then position, skip over positions that are lower, use ticket to take higher
 ;; Stick tickets in peer messenger group in single atom?

  (poll! [this]
    (.poll ^Subscription subscription ^FragmentHandler this fragment-limit-receiver))
  (set-heartbeat-peers! [this expected-peers]
    (println "Expected peers now" expected-peers)
    (set! peers expected-peers))
  (ready? [this]
    (assert (or ready (not= ready-peers peers)))
    (println "ready" ready "ready-peers" ready-peers peers)
    ready)
  (set-replica-version! [this new-replica-version]
    (set! ready false)


  (get-ticket [messenger {:keys [dst-task-id src-peer-id] :as sub}]
    (get-in @ticket-counters [replica-version [src-peer-id dst-task-id]]))

  (set-replica-version! [messenger rv]
    (assert (or (nil? replica-version) (> rv replica-version)) [rv replica-version])
