     (le/declare ch exchange-name "fanout")
     (lq/bind ch queue-name exchange-name)
   
-    {:ch ch
-     :consumer-tag (lc/subscribe ch queue-name handler)}))
+    (map->BroadcastConsumer {:ch ch
+                             :consumer-tag (lc/subscribe ch queue-name handler)})))
 
 (defn stop-broadcast-consumer! 
   "Shuts down the broadcast consumer. Returns nil."

(defn- bucket-consumer-shutdown-handler!
  "Cleans up resources on consumer shutdown based on observing changes to
   the consumer state."
  [old-state {:keys [instance-id status active empty-signal ch consumer-tag]}]

  (cond
    ;; Let the shutdown thread know (via blockingqueue) when all the active

  [state-atom & {:keys [force-stop]}]

    (loop []
      (let [{:keys [status ^BlockingQueue empty-signal]} (swap! state-atom stop-swap force-stop)]
        (when (not= status :stopped)
          (log/debugf "[%s] waiting for empty signal" (:instance-id @state-atom))
          (.take empty-signal) ;; wait for a message indicating we should try again
