在 Pedestal 中实施 SSE，每个用户使用单独的通道

Question

我正在开发一个曾经使用 websockets 的应用程序，但由于它们的扩展问题，我们决定用服务器发送事件 (SSE) 替换它们。

我们的目标是为每个成功登录的用户创建一个 SSE 通道，以便能够在用户注销或会话超时后关闭该通道，以免在给定用户登录后仍打开未使用的通道而消耗不必要的资源登出。我们将这些用户通道存储在原子映射注册表中，并希望访问特定登录用户的通道（通过用作注册表中的密钥的会话令牌进行识别）以将消息发送到前端。

我们最初使用了 io.pedestal.http.sse 命名空间，但它的问题是它期望处理程序接受“默认”通道作为路由处理程序输入参数，如果我尝试使用单独的用户通道，则消息未收到，例如在 Postman 中，我得到空响应，仅当您依赖处理程序作为输入参数获取的“默认”通道时，响应才包含预期消息：

(defn sse-handler [channel ctx] ...)

这是使用 start-event-stream 的路由，它返回一个拦截器，每当必须发送消息时，该拦截器就会使用处理程序：

(def sse-route ["/rest/sse" :get (sse/start-event-stream sse-handler)
    :route-name :sse])

这是处理程序：

(defn sse-handler [channel ctx]
  (doseq [[id {:keys [name data] :as payload}] @LIVE-NOTIFICATIONS]
    (when-not (chan/closed? channel)
      (async/>!! channel payload)
      (swap! LIVE-NOTIFICATIONS dissoc id)
      (Thread/sleep 1000)))
  (async/close! channel))

我的问题是：是否可以使用个人用户通道而不是上面粗体输入的通道来实现这样的场景以及如何实现？我是否必须实现自己的自定义拦截器而不是使用 start-event-stream 或者我仍然可以以某种方式使用 io.pedestal.http.sse ns？或者，如果我必须使用输入参数通道，是否可以以某种方式访问它以在用户注销时关闭它？

Answer 1

考虑到通道是由服务器本身处理的，你尝试以不同的方式处理 SSE 通道怎么样？不是你。

创建通道映射，然后在 SSE 处理程序中使用此映射中的通道

(defn channel-map [] (atom {}))
(defn get-user-channel [user-id channel-map](get @channel-map user-id))

(defn sse-handler [channel ctx]
(let [user-id (:user-id ctx)
     user-channel (get-user-channel user-id (channel-map))]
 (doseq [[id {:keys [name data] :as payload}] @LIVE-NOTIFICATIONS]
   (when-not (chan/closed? user-channel)
     (async/>!! user-channel payload)
     (swap! LIVE-NOTIFICATIONS dissoc id)
     (Thread/sleep 1000))))) 
 

当您想向用户发送消息时，可以使用

async/>!!

将消息排队到用户的频道中。要关闭用户的频道，您只需在用户注销或发生会话超时时从频道地图中

dissoc

获取用户ID即可。