 
     (defn- dist-add []
       (swap! distributors 
-             #(conj % (let [conn (rmq/connect {:vhost "boofa"})
+             #(conj % (let [conn (rmq/connect {:vhost "boofa" :requested-heartbeat 1 :connection-timeout 1})
                             scheduler (Executors/newScheduledThreadPool 1)
                             buckets (->> (range 100) (map str) (into []))]
                         (start-bucket-distributor! conn "bucket-too" buckets scheduler {}))))

  (when (log/enabled? :debug)
    (log-peer-changes last-peers peers peer-id))

  (when (not= last-size new-size)

    (log/infof "[%s] detected %s consumer(s), using bucket partition size of %s (was %s)" 
              peer-id (count peers) new-size last-size)

    (stop-bucket-consumer! consumer)

    (let [start-fn #(try
                      (start-bucket-consumer! conn queue-name new-size peer-id consumer true)
                      (catch Exception e
                        (log/errorf e "[%s] bucket-consumer start failed" peer-id)
                        nil))]
      (loop []
        (let [result (start-fn)]
          result
          (do 
            (interruptible-sleep 2000 peer-id)
            (log/infof "[%s] retrying bucket-consumer startup" peer-id)
            (recur)))))))

(defn peer-id-old []
  (let [hostname (-> (java.net.InetAddress/getLocalHost) .getHostName)

  (release! consumer (buckets! consumer))
  )
  
