http-kit sse

sse.clj

(ns something.core.sse
  (:require [charred.api :as json]
            [hiccup2.core :as hi]
            [org.httpkit.server :as httpd]
            [clojure.tools.logging :as log]))

(def ^:dynamic *report-ch* nil)

(declare sse!)

(defn sse-body [data]
  (str "data: " (json/write-json-str data) "\n\n"))

(defn do-with-sse [req cb]
  (httpd/as-channel req
    {:on-open (fn [ch]
                (httpd/send! ch
                  {:status  200
                   :headers {"Cache-Control" "no-store"
                             "Content-Type"  "text/event-stream"
                             "Connection"    "keep-alive"}
                   :body    (sse-body {:type :ping})}
                  false)
                (binding [*report-ch* ch]
                  (try
                    (cb)
                    (catch Exception e
                      (prn e)
                      (sse! :append [:article
                                     "Error:"
                                     [:p {:style "white-space: pre-wrap"} (pr-str e)]])))
                  (when (httpd/open? ch)
                    (sse! :end nil))))
     :on-close (fn [ch]
                 (prn "CLOSED" ch))}))

(defmacro with-sse [req & body]
  `(do-with-sse ~req (fn [] ~@body)))

(defn sse! [event html]
  (assert (contains? #{:start :redirect :append :replace :end} event))
  (if-not *report-ch*
    (log/warn "SSE cannot be sent" {:event event :html html})
    (httpd/send! *report-ch*
      {:body (sse-body {:type event
                        :html (hi/html html)})}
      ;; when event is redirect, then we close connection for sure
      (contains? #{:redirect :end} event))))

;;; Usage

(comment
  (defn handler [req]
    (with-sse req
      (sse! :start [:h1 "hello"])
      (Thread/sleep 1000)
      (sse! :append [:p "How do you do?"]))))