Clojure+Pedestal实现SSE返回空响应，请求技术排查

问题描述 #

基于Clojure、Pedestal和Jetty的Web应用中实现Server-Sent Events（SSE）时，出现以下异常：

• 后端打印显示通道已打开，调用io.pedestal.http.sse/send-event返回true
• 前端浏览器JavaScript无控制台日志输出，未接收任何数据
• Postman测试SSE连接成功，但响应显示为(empty)

后端原代码 #

(def SSE-REGISTRY (atom {}))
 
(defn get-user-channel [token]    
  (get @SSE-REGISTRY token))
 
(defn test-print [channel]    
  (println "channel:" channel) ;; channel:#object[clojure.core.async.impl.chan...]   
  (println "channel opened:" (not (chan/closed? channel))) ;; channel opened: true  
  (println "sent-event:" 
  (sse/send-event channel "status"
    (json/write-str {:id 1
                     :workflowId 3
                     :status :GOOD})) ;; sent-event: true
 
(defn send-sse-msg [name data id]   
  (when-let [sse-channels (vals @SSE-REGISTRY)]
    (doseq [channel sse-channels] 
      (when-not (chan/closed? channel)
        (test-print channel)
        (sse/send-event channel name data id)))))

(def sse-route ["/rest/sse" :get 
                 (sse/start-event-stream send-sse-msg 60 100) 
                 :route-name :sse])
 
(defn create-sse-channel []   
   (async/chan (async/sliding-buffer 100)))

前端原代码 #

const protocol = window.location.protocol;
const host = window.location.port; 
const sseUrl = protocol + '//'+ host + '/rest/sse';
const RUN_BUTTON_STATUS_TIME = 2000;
const sseInitOptionsMap = {
                            headers: {
                                       'Content-Type': 'text/event-stream; charset=utf-8',
                                       'Connection': 'keep-alive', 
                                       'Cache-Control': 'no-cache' 
                                      },
                            withCredentials: true,
                            https: {rejectUnauthorized: true}
                          };

export const eventSource = new EventSource(sseUrl, sseInitOptionsMap);
eventSource.addEventListener('integrationStatus', sendStatusHandler);
eventSource.addEventListener('stopAction', sendStopActionHandler);

eventSource.onopen = (e) => {
  console.log("SSE connection opened:" + e);
};

eventSource.onerror = (e) => {
  console.log("error:" + e);
  if (e.readyState == EventSource.CLOSED) {
      console.log("connection closed:" + e);
  } else {
    eventSource.close();
    console.log("SSE connection closed:" + e);
  }
};

export function sendStatusHandler(event) {
  const data = JSON.parse(event.data);
  console.log("data:" + data);
  let id = data.id;
  let workflowId = data.workflowId;
  let status = data.status;
  cljsDisp("set-global-operation-status", id, workflowId, status);
  configurtorActionRunButtonStatus(id, workflowId, status);
  setTimeout(cljsDisp("get-last-run-action-data", id, workflowId, status),
             300);
}

function configurtorActionRunButtonStatus (id, workflowId, status) {
  if (status === "GOOD") {
    showStatus(id, workflowId, true);
  } else if (status === "BAD") {
    showStatus(id, workflowId, true);
    cljsDisp("configurator-error-message-show", id, workflowId, true);
  } else {
    cljsDisp("configurator-action-run-button-status-visible?", id, workflowId, false);
  }
}

export function sendStopWorkflowHandler(event) {
  const data = JSON.parse(event.data);
  console.log("data:" + data); // prints nothing
  let workflowId = data.workflowId;
  cljsDisp("stop-action-by-sse-msg", workflowId);
}

function showStatus(integrationId, operationId, showStatus) {
  setTimeout(cljsDisp("configurator-action-run-button-status-visible?", 
                      integrationId, operationId, showStatus),
             RUN_BUTTON_STATUS_TIME);
}

export function closeSse() {
  if (eventSource.readyState != eventSource.CLOSED) {
    eventSource.close();
  }
}

核心问题排查与修复 #

后端问题修复 #

1. 事件名称不匹配
   后端发送的事件名称是"status"，但前端监听的是"integrationStatus"，完全不对应。需统一事件名称，比如后端改为发送"integrationStatus"。

2. SSE通道未注册到注册表
   sse/start-event-stream的回调函数需要接收通道作为第一个参数（Pedestal的start-event-stream会在连接建立时将新创建的通道传入回调），原代码的send-sse-msg未处理通道注册逻辑，导致SSE-REGISTRY始终为空。修正回调函数：

   ;; 修正send-sse-msg，添加channel参数并注册到注册表（示例用请求头token作为键，需根据实际认证逻辑调整）
   (defn send-sse-msg [channel name data id]   
     (let [token (get-in (:request channel) [:headers "x-user-token"])]
       (swap! SSE-REGISTRY assoc token channel))
     (when-not (chan/closed? channel)
       (test-print channel)
       (sse/send-event channel name data id)))
   

3. 语法错误修复
   test-print函数缺少闭合括号，导致代码执行异常，修正后：

   (defn test-print [channel]    
     (println "channel:" channel)
     (println "channel opened:" (not (chan/closed? channel)))
     (println "sent-event:" 
              (sse/send-event channel "integrationStatus"
                              (json/write-str {:id 1
                                               :workflowId 3
                                               :status :GOOD}))))
   

4. 启用流式响应配置
   确保Pedestal的HTTP服务配置中开启:streaming? true，否则SSE响应会被缓冲，无法实时推送：

   (def service {:env :prod
                 ::http/routes routes
                 ::http/type :jetty
                 ::http/port 8080
                 ::http/streaming? true ;; 必须开启
                 })
   

前端问题修复 #

1. 修正URL拼接错误
   原代码只取了端口window.location.port，未包含主机名，导致URL无效。修正为：

   const sseUrl = `${window.location.protocol}//${window.location.host}/rest/sse`;
   

2. 移除不合法的EventSource初始化参数
   标准EventSource API不支持自定义headers和https配置，这些参数会被浏览器忽略或导致初始化失败。保留合法参数：

   const sseInitOptionsMap = {
     withCredentials: true
   };
   

3. 统一事件监听名称
   确保监听的事件名称与后端发送的一致，比如后端发送"integrationStatus"，前端保持对应：

   eventSource.addEventListener('integrationStatus', sendStatusHandler);
   

内容的提问来源于stack exchange，提问作者Rosen Hristov