   "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])
+  (when (and @open (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)
+          (do (wamp/emit-event! topic (dissoc event :topic) sess-id)
               (recur))
           (unsubscribe state sess-id topic)))
       (api/watch-sources dirreader topic :none sourcesc))))

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

                            :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)))
