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

  "Given the first item in the return value of `make-app` function,
  this function will unwatch the current open watches to the
  DirectoryReader. The app should not and cannot be used afterwards."
  [{:keys [open channels] :as state}]
  (info "Stopping the reader application ...")
  (reset! open false)
  (doseq [[sess-id topic-chans] @channels
          [topic _] topic-chans]
    (unsubscribe state sess-id topic))
  (doseq [[sess-id _] @wamp/client-channels]
    (wamp/close-channel sess-id))
  (info "Stopped the reader application."))


;;; Exposing a DirectoryReader as a Websocket service.

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


(defn- unsubscribe
