                                     (catch Throwable e (fatal e))))]
     {:conn conn :subscription subscription :subscriber-fut subscriber-fut}))
 
-
-;; FIXME, race condition 
+(defrecord TrackedPublication [publication last-used])
+
 (defn get-publication [messenger channel]
+  ;; FIXME, race condition may cause two publications to be created
+  ;; same goes for operation/peer-link
   (if-let [pub (get @(:publications messenger) channel)]
-    pub
+    (do 
+      (reset! (:last-used pub) (System/currentTimeMillis))
+      (:publication pub))
     (let [conn (Aeron/connect (.errorHandler (Aeron$Context.) no-op-error-handler))
           pub (.addPublication conn channel send-stream-id)]
-      (do (swap! (:publications messenger) assoc channel pub)
+      (do (swap! (:publications messenger) assoc channel (->TrackedPublication pub (atom (System/currentTimeMillis))))
           (swap! (:connections messenger) assoc channel conn)
           pub))))
 

             :release-ch release-ch
             :retry-ch retry-ch)))

  (stop [{:keys [aeron resources release-ch] :as component}]
    (taoensso.timbre/info "Stopping Aeron")
    (try 
      ;;; TODO; could handle the inbound-ch in a similar way - rather than using messenger-buffer
      (close! (:release-ch component))
      (close! (:retry-ch component))
      (catch Throwable e (fatal e)))

    (assoc component
           :send-idle-strategy nil 
           :short-circuitable? nil

                 Available ports can be configured in the peer-config.
                 e.g. {:onyx.messaging/peer-ports [40000, 40002],
                 :onyx.messaging/peer-port-range [40200 40260]}"))))


;; FIXME: gc'ing publications could be racy if a publication is grabbed 
;; just as it is being gc'd - though it is unlikely
(defn gc-publications [publications connections opts]
  (let [interval (arg-or-default :onyx.messaging/peer-link-gc-interval opts)
        idle (arg-or-default :onyx.messaging/peer-link-idle-timeout opts)]
    (loop []
      (try
        (Thread/sleep interval)
        (let [t (System/currentTimeMillis)
              snapshot @publications
              to-remove (map first 
                             (filter (fn [[k v]] (>= (- t @(:last-used v)) idle)) 
                                     snapshot))]
          (doseq [k to-remove]
            (let [pub (:publication (snapshot k))
                  conn (@connections k)] 
              (swap! publications dissoc k)
              (swap! connections dissoc k)
              (.close pub)
              (.close conn))))
        (catch InterruptedException e
          (throw e))
        (catch Throwable e
          (fatal e)))
      (recur))))

(defrecord AeronPeerGroup [opts publications connections compress-f decompress-f send-idle-strategy]
  component/Lifecycle
