我正在使用Clojure编写一个应用程序服务器,客户端将使用ClojureScript。
我想找到一种高效、符合惯例的方法,以实时事件形式从服务器推送数据到客户端,最好是使用以下某种组合:
- http-kit
- core.async
- Ring
(但我也可以接受其他可能性)
有人能提供一个好的示例/方法来完成这个任务吗?
我正在使用Clojure编写一个应用程序服务器,客户端将使用ClojureScript。
我想找到一种高效、符合惯例的方法,以实时事件形式从服务器推送数据到客户端,最好是使用以下某种组合:
(但我也可以接受其他可能性)
有人能提供一个好的示例/方法来完成这个任务吗?
wrap-ring-handler
函数简单地包装现有的处理程序。最近我一直在尝试使用库 Chord,我非常喜欢它。
它提供了一个小的 core.async 包装器,围绕着 http-kit 中的 Websocket 支持。
从 Github 页面上看:
在服务器上
(:require [chord.http-kit :refer [with-channel]]
[clojure.core.async :refer [<! >! put! close! go]])
(defn your-handler [req]
(with-channel req ws-ch
(go
(let [{:keys [message]} (<! ws-ch)]
(println "Message received:" message)
(>! ws-ch "Hello client from server!")
(close! ws-ch)))))
在客户端
(:require [chord.client :refer [ws-ch]]
[cljs.core.async :refer [<! >! put! close!]])
(:require-macros [cljs.core.async.macros :refer [go]])
(go
(let [ws (<! (ws-ch "ws://localhost:3000/ws"))]
(>! ws "Hello server from client!")))
我认为它仍处于早期阶段 - 它还不能处理断开连接的情况。
core.async
相结合来实现服务器发送事件(SSE),并且运行得非常好。new EventSource(SOME_URI);
中传递任何自定义标头。)(ns chat-service.service
(:require [clojure.set :as set]
[clojure.core.async :as async :refer [<!! >!! <! >!]]
[cheshire.core :as json]
[io.pedestal.service.http :as bootstrap]
[io.pedestal.service.log :as log]
[io.pedestal.service.http.route :as route]
[io.pedestal.service.http.sse :as sse]
[io.pedestal.service.http.route.definition :refer [defroutes]]))
(def ^{:private true :doc "Formatting opts"} json-opts {:date-format "MMM dd, yyyy HH:mm:ss Z"})
(def ^{:private true :doc "Users to notification channels"} subscribers->notifications (atom {}))
;; private helper functions
(def ^:private generate-id #(.toString (java.util.UUID/randomUUID)))
(defn- sse-msg [event msg-data]
{:event event :msg msg-data})
;; service functions
(defn- remove-subscriber
"Removes transport channel from atom subscribers->notifications and tears down
SSE connection."
[transport-channel context]
(let [subscriber (get (set/map-invert @subscribers->notifications) transport-channel)]
(log/info :msg (str "Removing SSE connection for subscriber with ID : " subscriber))
(swap! subscribers->notifications dissoc subscriber)
(sse/end-event-stream context)))
(defn send-event
"Sends updates via SSE connection, takes also transport channel to close it
in case of the exception."
[transport-channel context {:keys [event msg]}]
(try
(log/info :msg "calling event sending fn")
(sse/send-event context event (json/generate-string msg json-opts))
(catch java.io.IOException ioe
(async/close! transport-channel))))
(defn create-transport-channel
"Creates transport channel with receiving end pushing updates to SSE connection.
Associates this transport channel in atom subscribers->notifications under random
generated UUID."
[context]
(let [temporary-id (generate-id)
channel (async/chan)]
(swap! subscribers->notifications assoc temporary-id channel)
(async/go-loop []
(when-let [payload (<! channel)]
(send-event channel context payload)
(recur))
(remove-subscriber channel context))
(async/put! channel (sse-msg "eventsourceVerification"
{:handshakeToken temporary-id}))))
(defn subscribe
"Subscribes anonymous user to SSE connection. Transport channel with timeout set up
will be created for pushing any new data to this connection."
[context]
(create-transport-channel context))
(defroutes routes
[[["/notifications/chat"
{:get [::subscribe (sse/start-event-stream subscribe)]}]]])
(def service {:env :prod
::bootstrap/routes routes
::bootstrap/resource-path "/public"
::bootstrap/type :jetty
::bootstrap/port 8081})
我遇到的一个问题是Pedestal默认处理断开的SSE连接的方式。
因为有定期的心跳作业,所以每当连接断开时,并且您没有调用 end-event-stream 上下文,它就会记录异常。
我希望有一种方法可以禁用/调整此行为,或者至少提供自己的拆除函数,每当心跳作业由于 EofException 失败时将调用该函数。