       (do (println e)
           message))))
 
+(defn connect! [room-name user-name]
+  (fn [conn]
+    ;; handler for disconnects
+    (s/on-closed conn #(remove-user! room-name user-name))
+    
+    ;; chatrooms bus -> websocket
+    (s/connect (bus/subscribe chatrooms room-name) conn)
+    
+    ;; websocket -> chatrooms bus (kind of doseq for streams)
+    (s/consume
+     #(bus/publish! chatrooms room-name (pack-room-shift! user-name room-name %))
+     (s/throttle 10 conn))))
+
 ;; HELPERS
 
 (defn filter-empty-vals [params]

    (let [room (site/get-room @site room-name)]
      (f room))))

(defn with-websocket-check [req exp]
  (d/let-flow [conn (d/catch (http/websocket-connection req) (fn [_] nil))]
      (if-not conn
        #'json-non-websocket-response)
      (exp conn)))

;;; HANDLERS


    (with-websocket-check req #(s/connect (bus/subscribe chatrooms room-name) %))
    
    :else
    (run-checks
     (-> (connect! room-name user-name)
         (with-websocket-check req)
         (with-room-check room-name)
         (with-user-check room-name user-name)))))

(defn reconnect-handler [req room-name user-name]
  (println "reconnecting" user-name "to" room-name)
  (if (is-observer? user-name)
    (with-websocket-check req #(s/connect (bus/subscribe chatrooms room-name) %))    
    (run-checks
     (-> (connect! room-name user-name)
         (with-websocket-check req)
         (with-room-check room-name)
         (with-user-check room-name user-name)))))

(defn remove-user-handler [req]
  (let [user-name  (get-in req [:body "user-name"])
