如何用NestJS反向代理SSE请求？解决UI不更新问题

我们有一套包含Node.js后端与Vue前端的外部Web应用，需部署在自研反向代理（非Nginx）之后以绕过特定内部逻辑。该应用采用SSE（Server-Sent Events）更新UI，但目前NestJS仅实现了普通HTTP请求代理，SSE请求代理存在异常——推送请求在浏览器中显示完成，但前端UI并未收到更新。

已实现代码 #

SSE控制器代码 #

@Sse('/rest/push')
async proxyPushMessages(@Res({ passthrough: true }) response: FastifyReply, @Req() request: any) {
  const sseStream = this.sseProxyService.proxySSE(request);
  response.header('Content-Type', 'text/event-stream');
  response.header('Cache-Control', 'no-cache');
  response.header('Connection', 'keep-alive');
  sseStream.subscribe(
    (messageEvent) => {
      return response.send(`data: ${messageEvent.data}`);
    },
    (error) => {
      console.error('SSE Error:', error);
      return response.send(`data: ${error}`);
    },
    () => {
      return response.send(`data: ${'SSE stream closed'}`);
    },
  );
}

SSE代理服务代码 #

import { map, Observable } from 'rxjs';
import { HttpService } from '@nestjs/axios';
import { FastifyRequest, Injectable } from '@transmit-security/bindid-services-core-service';
import { AxiosRequestConfig } from 'axios';
const baseWorkflowsUrl = process.env.WORKFLOWS_SERVER_URL || 'http://localhost:5678';

@Injectable()
export class SSEProxyService {
  constructor(private httpService: HttpService) {}
  private buildUrl(url: string): string {
    return `${baseWorkflowsUrl}${url}`;
  }
  proxySSE(req: FastifyRequest): Observable<MessageEvent> {
    const { url, method, headers } = req;
    const reqObject: AxiosRequestConfig = {
      url: this.buildUrl(url),
      withCredentials: true,
      responseType: 'text',
      method,
      headers: headers as any,
    };
    return this.httpService.request(reqObject).pipe(map((response) => new MessageEvent('message', response.data)));
  }
}

问题根源与修复方案 #

1. Axios未处理流式响应 #

默认Axios会等待整个响应完成后才返回结果，但SSE是流式分块传输，需要实时接收每一块数据。当前代码中responseType: 'text'会导致Axios缓存所有数据，直到连接关闭才一次性返回，无法实现SSE的实时推送。

修复：将responseType改为'stream'，并通过监听流的data事件创建Observable，实时转发每一块数据：

// 修改SSEProxyService的proxySSE方法
proxySSE(req: FastifyRequest): Observable<string> {
  const { url, method, headers } = req;
  const reqObject: AxiosRequestConfig = {
    url: this.buildUrl(url),
    withCredentials: true,
    responseType: 'stream', // 改为流式响应
    method,
    headers: { ...headers, 'Accept': 'text/event-stream' }, // 明确指定接收SSE类型
  };

  return new Observable((observer) => {
    this.httpService.request(reqObject).subscribe({
      next: (response) => {
        // 监听数据流的每一块数据
        response.data.on('data', (chunk: Buffer) => {
          observer.next(chunk.toString());
        });
        // 监听流关闭
        response.data.on('end', () => {
          observer.complete();
        });
        // 监听流错误
        response.data.on('error', (err) => {
          observer.error(err);
        });
      },
      error: (err) => {
        observer.error(err);
      },
    });
  });
}

2. SSE消息格式不规范 #

SSE协议要求每条消息必须以**两个换行符（\n\n）**结尾，浏览器才会识别为完整的消息并触发message事件。当前代码仅发送data: ${messageEvent.data}，缺少结尾的换行，导致前端无法解析消息。

修复：在控制器中发送消息时添加结尾换行，若上游返回标准SSE格式则直接转发：

// 修改控制器的subscribe回调
sseStream.subscribe(
  (chunk) => {
    // 直接转发原始chunk（已包含SSE格式），无需额外拼接data:
    response.send(chunk);
  },
  (error) => {
    console.error('SSE Error:', error);
    response.send(`data: ${JSON.stringify(error.message)}\n\n`);
  },
  () => {
    response.send(`data: SSE stream closed\n\n`);
    response.end();
  },
);

3. Fastify响应的流式处理优化 #

Fastify处理流式数据时，需确保响应处于分块传输模式，可显式添加响应头或直接操作底层响应：

// 在控制器开头添加分块传输头
response.header('Transfer-Encoding', 'chunked');
// 或使用raw直接写入数据
response.raw.write(chunk);

4. 请求头传递问题 #

确保上游服务的Accept头被设置为text/event-stream，同时过滤掉Content-Length这类会干扰流式传输的请求头。

最终验证 #

修改完成后检查浏览器网络请求：

• 响应头需包含Content-Type: text/event-stream、Cache-Control: no-cache、Connection: keep-alive、Transfer-Encoding: chunked
• 响应体应为分块传输的SSE消息，每条消息以\n\n结尾
• 前端EventSource实例应能正常触发message事件，更新UI

内容的提问来源于stack exchange，提问作者Matan Tubul