websockets with core.async and http-kit

anonymous-gist.clj

(defn build-routes
  [{:keys [bus queues] :as app-data}]
  (routes
   (GET "/data" req
        ; get an http-kit websocket channel for this request
        (http/with-channel req req-ch
          ; make a channel for subscriptions to the data updated event bus
          (let [metrics (async/chan)]
            ; subscribed to all messages in the updated topic
            (async/sub bus :updated metrics)
            ; when the http-kit channel closes, also close the event bus subscription
            (http/on-close req-ch (partial async/close! metrics))

            ; loop in a go-routine
            (async/go-loop []
              ; read (<!!) will block until there is a message to read
              (async/<!! metrics)

              ; send whatever is the current state of app-data down the channel
              (http/send! req-ch (json/write-str @queues))
              ; only loop if the websocket channel is still open
              (when (http/open? req-ch) (recur))))))



   (route/resources "/")
   (route/not-found "Not Found")))

(defrecord AppRoutes []
  component/Lifecycle

  (start [this]
    (assoc this
      :routes (build-routes (:data this))))

  (stop [this] this))