     (debug "Cannot subscribe" sess-id "to" topic "- reader is closing or already subscribed.")))
 
 
+(defn- handle-open
+  "Handles an opened connection. This will register the session, in
+  order to close it whenever the app is stopped."
+  [{:keys [sessions] :as state} sess-id]
+  (swap! sessions conj sess-id))
+
+
 (defn- handle-close
   "Handles a closed connection. This will automatically unsubscribe
   the channel from all topics in the DirectoryReader."
-  [{:keys [channels] :as state} sess-id status]
+  [{:keys [channels sessions] :as state} sess-id status]
   (debug "Handling close for" sess-id "given status" status)
   (doseq [[topic _] (get @channels sess-id)]
-    (unsubscribe state sess-id topic)))
+    (unsubscribe state sess-id topic))
+  (swap! sessions disj sess-id))
 
 
 (defn- send-sources

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

  for the `stop-app` function, and the second is the ring app
  function."
  [directory-reader & {:keys [subscribe-buffer] :as config}]
  (let [state (State. directory-reader config (atom {}) (atom nil) (atom true) (atom #{}))
        app (make-app* state)]
    [state app]))

