   the server and unsubscribe every connection in the DirectoryReader."
   [{:keys [stop-fn open channels] :as state}]
   (reset! open false)
+  (doseq [[sess-id topic-chans] @channels
+          [topic _] topic-chans]
+    (unsubscribe state sess-id topic))
   (@stop-fn :timeout 100))

  "Subscribe the given session to the given topic. This starts a
  go-loop, reading from a sliding-buffer channel."
  [{:keys [dirreader config channels open] :as state} sess-id topic]
  (prn 'SUB? channels sess-id topic)
  (when-not (get-in @channels [sess-id topic])
    (let [sourcesc (async/chan (async/sliding-buffer (get config :subscribe-buffer 100)))]
      (swap! channels assoc-in [sess-id topic] sourcesc)
      (async/go-loop []
        (if-let [event (<! sourcesc)]
          (do (wamp/emit-event! topic event sess-id)
              (recur))
          (unsubscribe state sess-id topic)))
      (api/watch-sources dirreader topic :none sourcesc))))

                            :on-subscribe {"*" true
                                           :on-after (partial subscribe state)}
                            :on-unsubscribe (partial unsubscribe state)
                            :on-close (partial handle-close state)
                            :on-call {"sources" (partial send-sources state)}})]
              (when (seq topic)
                (wamp/topic-subscribe topic sess-id)
                (subscribe state sess-id topic)))
