Angular中如何将Server-Sent Events的EventSource事件流封装为Observable并解决类型错误问题

我来帮你梳理下问题所在，以及给出可行的封装方案：

为什么你的代码不工作？ #

1. 没有触发订阅者通知：你创建了EventSource实例，但从未调用subscriber.next()向订阅者发送任何数据，所以订阅回调自然不会执行。
2. 返回值类型错误：Observable的构造函数回调需要返回TeardownLogic（清理函数、Subscription或void），而你返回了sse.readyState（数字类型），这就导致了类型不匹配的错误。

可行的封装方案 #

根据你的需求——将EventSource的连接逻辑单独封装，待连接成功后再执行后续业务逻辑——这里有两种常见的实现方式：

方案1：仅封装连接成功事件（返回EventSource实例）

这个方案会在EventSource连接成功（触发onopen）时，将实例传递给订阅者，你可以基于这个实例再处理后续的消息事件：

openEventSource(url: string): Observable<EventSource> {
  return new Observable<EventSource>(subscriber => {
    const sse = new EventSource(url);

    // 连接成功时，把EventSource实例发送给订阅者
    sse.onopen = () => {
      subscriber.next(sse);
      // 注意：不要调用subscriber.complete()，因为EventSource是长连接，需要保持订阅状态
    };

    // 连接出错时，通知订阅者并关闭连接
    sse.onerror = (error) => {
      subscriber.error(error);
      sse.close();
    };

    // 返回清理函数：当订阅取消时关闭EventSource
    return () => {
      if (sse.readyState === EventSource.OPEN) {
        sse.close();
      }
    };
  });
}

使用方式：

this.eventStreamHandlerServ.openEventSource('http://localhost:3000/sse/event')
  .pipe(
    switchMap(sse => {
      // 连接成功后，创建Observable处理后续的message事件
      return new Observable<MessageEvent<any>>(messageSubscriber => {
        const handleMessage = (e: MessageEvent) => messageSubscriber.next(e);
        const handleError = (e: ErrorEvent) => messageSubscriber.error(e);

        sse.onmessage = handleMessage;
        sse.onerror = handleError;

        // 清理函数：取消事件监听
        return () => {
          sse.onmessage = null;
          sse.onerror = null;
        };
      });
    })
  )
  .subscribe(
    message => console.log('收到SSE消息:', message),
    err => console.error('SSE错误:', err)
  );

方案2：封装完整的EventSource事件流（连接、消息、错误）

如果你希望把所有EventSource事件都整合到一个Observable中，可以用这种方式，通过区分事件类型来处理：

openEventSourceFull(url: string): Observable<
  { type: 'open' } | 
  { type: 'message', data: MessageEvent<any> } | 
  { type: 'error', error: ErrorEvent }
> {
  return new Observable(subscriber => {
    const sse = new EventSource(url);

    sse.onopen = () => subscriber.next({ type: 'open' });
    sse.onmessage = (e) => subscriber.next({ type: 'message', data: e });
    sse.onerror = (error) => subscriber.next({ type: 'error', error });

    // 清理函数：关闭连接
    return () => {
      if (sse.readyState === EventSource.OPEN) {
        sse.close();
      }
    };
  });
}

使用方式：

this.eventStreamHandlerServ.openEventSourceFull('http://localhost:3000/sse/event')
  .subscribe(event => {
    switch(event.type) {
      case 'open':
        console.log('SSE连接成功');
        // 在这里执行连接后的初始化逻辑
        break;
      case 'message':
        console.log('收到消息:', event.data);
        break;
      case 'error':
        console.error('连接错误:', event.error);
        break;
    }
  });

关键注意点 #

• 清理逻辑：一定要在Observable的清理函数中关闭EventSource，避免内存泄漏。
• 错误处理：根据业务需求决定是用subscriber.error()（终止订阅）还是发送错误事件（保持订阅），如果是临时连接错误，EventSource会自动重连，此时用事件通知更合适。
• RxJS操作符：使用switchMap可以确保在连接变更时自动取消之前的订阅，避免多个EventSource实例同时存在。

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