   [{:keys [channels] :as state} sess-id status]
   (doseq [[topic _] (get @channels sess-id)]
     (unsubscribe state sess-id topic)))
+
+
+(defn- send-sources
+  "Sends the currently known sources to the caller."
+  [{:keys [dirreader open] :as state} topic publish?]
+  (if @open
+    (let [sources (api/sources dirreader topic)]
+      (if publish?
+        (do (doseq [source sources]
+              (wamp/emit-event! topic {:event :joined, :uri source} wamp/*call-sess-id*))
+            {:result (str "Published " (count sources) " sources as events.")})
+        {:result sources}))
+    {:error {:uri topic
+             :message "Server closing"
+             :description (str "The server is closing.")
+             :kill false}}))
 
 
 (defn- make-app

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


;;; Exposing a DirectoryReader as a Websocket service.

;; channels = (atom {"sess-id" {"topic" async.chan}})
(defrecord State [dirreader config channels stop-fn open])


