Angular 16中如何拦截WebSockets、SSE及ReadableStream的请求与响应？

一、WebSockets/SSE/ReadableStream 拦截实现案例 #

Angular 原生 HttpInterceptor 仅作用于 HttpClient 发起的请求，WebSocket、SSE、原生 Fetch 这类 HTML5 API 无法直接被拦截，需通过自定义封装模拟拦截逻辑：

1. WebSocket 拦截实现 #

封装专属 WebSocket 服务，在连接建立、消息收发环节手动加入拦截逻辑（如认证、日志、消息转换）：

@Injectable({ providedIn: 'root' })
export class WebSocketService {
  private socket: WebSocket | null = null;

  constructor(private authService: AuthService) {}

  connect(url: string): Observable<any> {
    // 请求拦截：注入认证参数
    const authUrl = `${url}?token=${this.authService.getAuthToken()}`;
    this.socket = new WebSocket(authUrl);

    return new Observable(observer => {
      // 响应拦截：统一解析消息格式
      this.socket!.onmessage = (event) => {
        try {
          const parsedData = JSON.parse(event.data);
          observer.next(parsedData);
        } catch (e) {
          observer.error('消息解析失败');
        }
      };

      this.socket!.onerror = (error) => observer.error(error);
      this.socket!.onclose = () => observer.complete();
    });
  }

  sendMessage(data: any): void {
    if (this.socket?.readyState === WebSocket.OPEN) {
      // 发送前拦截：统一格式化消息
      const formattedData = JSON.stringify(data);
      this.socket.send(formattedData);
    }
  }
}

2. SSE 拦截实现 #

封装 SSE 服务，结合自定义拦截器接口，模拟请求/响应拦截流程：

// 定义SSE拦截器接口
export interface SseInterceptor {
  beforeConnect(url: string, options: EventSourceInit): { url: string; options: EventSourceInit };
  afterMessageReceived(event: MessageEvent): MessageEvent;
}

// 实现认证拦截器
@Injectable()
export class AuthSseInterceptor implements SseInterceptor {
  constructor(private authService: AuthService) {}

  beforeConnect(url: string, options: EventSourceInit) {
    // 请求拦截：添加认证头
    options.headers = {
      ...options.headers,
      'Authorization': `Bearer ${this.authService.getAuthToken()}`
    };
    return { url, options };
  }

  afterMessageReceived(event: MessageEvent) {
    // 响应拦截：解析JSON消息
    event.data = JSON.parse(event.data);
    return event;
  }
}

// 核心SSE服务
@Injectable({ providedIn: 'root' })
export class SseService {
  constructor(
    private authSseInterceptor: AuthSseInterceptor
  ) {}

  getServerEvents(url: string): Observable<MessageEvent> {
    const interceptors: SseInterceptor[] = [this.authSseInterceptor];
    let finalUrl = url;
    let finalOptions: EventSourceInit = {};

    // 执行所有请求拦截器
    interceptors.forEach(interceptor => {
      const result = interceptor.beforeConnect(finalUrl, finalOptions);
      finalUrl = result.url;
      finalOptions = result.options;
    });

    const eventSource = new EventSource(finalUrl, finalOptions);

    return new Observable(observer => {
      eventSource.onmessage = (event) => {
        // 执行所有响应拦截器
        let processedEvent = event;
        interceptors.forEach(interceptor => {
          processedEvent = interceptor.afterMessageReceived(processedEvent);
        });
        observer.next(processedEvent);
      };

      eventSource.onerror = (error) => observer.error(error);
      eventSource.onclose = () => observer.complete();
    });
  }
}

3. ReadableStream 与 Fetch API 扩展 #

要让 Fetch 请求走 Angular 拦截链，有两种可行方案：

方案1：封装自定义 Fetch 服务，复用 HttpClient 拦截器

手动将 Fetch 请求转换为 HttpRequest，调用拦截器链后再转换回 Fetch 响应：

@Injectable({ providedIn: 'root' })
export class FetchService {
  constructor(
    private interceptors: HttpInterceptor[],
    private httpBackend: HttpBackend
  ) {}

  async fetch(input: RequestInfo, init?: RequestInit): Promise<Response> {
    // 转换为HttpClient请求格式
    const request = new HttpRequest(
      init?.method || 'GET',
      typeof input === 'string' ? input : input.url,
      init?.body,
      {
        headers: new HttpHeaders(init?.headers),
        responseType: 'blob'
      }
    );

    // 构建拦截器链
    let handler: HttpHandler = this.httpBackend;
    for (const interceptor of this.interceptors.reverse()) {
      handler = new InterceptorHandler(interceptor, handler);
    }

    // 执行请求并转换为Fetch响应
    const httpResponse = await handler.handle(request).toPromise();
    return new Response(httpResponse.body, {
      status: httpResponse.status,
      statusText: httpResponse.statusText,
      headers: new Headers(
        httpResponse.headers.keys().reduce((acc, key) => {
          acc[key] = httpResponse.headers.getAll(key);
          return acc;
        }, {} as Record<string, string[]>)
      )
    });
  }
}

// 辅助类：实现拦截器链处理
class InterceptorHandler implements HttpHandler {
  constructor(private interceptor: HttpInterceptor, private next: HttpHandler) {}

  handle(request: HttpRequest<any>): Observable<HttpEvent<any>> {
    return this.interceptor.intercept(request, this.next);
  }
}

使用时直接调用 this.fetchService.fetch(url, options) 替代原生 fetch，即可触发所有 Angular 拦截器。

方案2：全局重写 fetch 方法

通过 APP_INITIALIZER 替换全局 window.fetch，让所有原生 Fetch 请求自动走拦截链：

export function overrideGlobalFetch(fetchService: FetchService) {
  return () => {
    window.fetch = async (input, init) => {
      return fetchService.fetch(input, init);
    };
  };
}

// 在AppModule中配置
@NgModule({
  providers: [
    FetchService,
    {
      provide: APP_INITIALIZER,
      useFactory: overrideGlobalFetch,
      deps: [FetchService],
      multi: true
    }
  ]
})
export class AppModule {}

二、HttpClientModule 与 HTML5 API 结合的实践现状 #

Angular 官方未提供直接拦截 WebSocket、SSE、原生 Fetch 的支持，社区普遍采用自定义封装服务+模拟拦截逻辑的方案：

• 对 WebSocket/SSE：封装专属服务，在连接、消息环节手动加入认证、日志、格式转换等拦截操作；
• 对 Fetch API：通过自定义服务复用现有 HttpInterceptor 链，或全局重写 fetch 方法实现拦截。

内容的提问来源于stack exchange，提问作者Hari Prasad