reconnecting_websocket.clj

(defn re-ws-stream [url]
  (let [s (ms/stream 10)
        renew-connection
        (fn this
          ([] (this 0 (System/currentTimeMillis)))
          ([conn-ct prev-conn-ts]
           (-> (http/websocket-client url {:max-frame-payload 1e7 :max-frame-size 1e7})
               (md/chain 
                 (fn [conn]
                   (log/info "ws connected. connection count" conn-ct)
                   (let [conn-ts (System/currentTimeMillis)]
                     (ms/on-closed s (fn [] (ms/close! conn)))
                     (ms/on-closed conn
                       (fn []
                         (if (ms/closed? s)
                           (log/info "ws closed intentionally")
                           (do (when (= (quot prev-conn-ts 1000)
                                        (quot conn-ts 1000))
                                (log/info "attempted to reset more than once per second, waiting")
                                (Thread/sleep 2000))
                               (log/info "ws closed unintentionally")
                               (this (inc conn-ct) conn-ts)))))
                     (ms/put! s conn))))
               (md/catch Exception
                   (fn [e]
                     (println "error in ws connection")
                     (pst e))))))]

    (renew-connection)
    s))