 
   (get-ticket [messenger {:keys [dst-task-id src-peer-id] :as sub}]
     (get-in @ticket-counters [replica-version [src-peer-id dst-task-id]]))
-
-  (add-publication [messenger pub-info]
-    (set! publications 
-          (update-in publications
-                     [(:dst-task-id pub-info) (:slot-id pub-info)]
-                     (fn [pbs] 
-                       (assert (= id (:src-peer-id pub-info)) [id (:src-peer-id pub-info)])
-                       (conj (or pbs []) 
-                             (pub/start (new-publication messenger messenger-group pub-info))))))
-
-    messenger)
-
-  (remove-publication [messenger pub-info]
-    (println "Remove publication:" pub-info)
-    (set! publications (update-in publications 
-                                  [(:dst-task-id pub-info) (:slot-id pub-info)] 
-                                  remove-from-publications 
-                                  pub-info))
-    messenger)
 
   (set-replica-version! [messenger rv]
     (assert (or (nil? replica-version) (> rv replica-version)) [rv replica-version])

  (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)

               (stream-id job-id dst-task-id slot-id site)
               nil nil nil))

;; 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?
