           broadcast! #(send-broadcast! conn broadcast-exchange peer-id %)
 
           ; core stuff that gets passed around
-          distributor {:options options
-                       :default-buckets default-buckets
-                       :peer-id peer-id 
-                       :state-atom state-atom
-                       :broadcast! broadcast!}
+          distributor (map->RabbitBucketDistributor {:options options
+                                                     :default-buckets default-buckets
+                                                     :peer-id peer-id 
+                                                     :state-atom state-atom
+                                                     :broadcast! broadcast!})
 
           broadcast-consumer (start-broadcast-consumer! conn broadcast-exchange
                                                         #(handle-broadcast! distributor %1 %2))


  (dist-add)
  (dist-remove)
  (clojure.pprint/pprint distributors))

(defprotocol BucketDistributor "A mechanism for coordinated distribution of
                                hash buckets."

                                 partition-delay partition-period partition-units)))))

(defn stop-bucket-distributor! 
  [{:keys [broadcast! broadcast-consumer ^Future peers-future ^Future partition-future peer-id state-atom]}]

  (log/infof "[%s] stopping distributor" peer-id)

