Angular14+RxJS7：如何将WebSocket的Next消息封装为可订阅Observable

核心思路 #

服务内部统一管理WebSocket的连接、错误和完成事件，仅将消息流（Next事件）暴露给组件，组件无需处理错误逻辑，专注业务即可。

服务端代码修改 #

import { Injectable } from '@angular/core';
import { webSocket, WebSocketSubject } from 'rxjs/webSocket';
import { Observable, tap, catchError, share, EMPTY } from 'rxjs';

@Injectable({ providedIn: 'root' })
export class WebsocketService {
  private ws$: WebSocketSubject<any> = webSocket('ws://127.0.0.1:2015');
  
  // 预封装消息流：拦截错误、共享连接、仅传递有效消息
  private messageStream$: Observable<any> = this.ws$.pipe(
    // 服务内部记录消息日志
    tap(msg => console.log('service received message: ' + JSON.stringify(msg))),
    // 捕获并内部处理错误，不传递给组件
    catchError(err => {
      console.error('websocket error:', err);
      // 返回空流维持订阅，也可根据需求添加重连逻辑
      return EMPTY;
    }),
    // 多组件订阅共享同一个WebSocket连接，避免重复建立连接
    share()
  );

  public connect(): void {
    // 启动WebSocket连接，统一处理错误和关闭事件
    this.ws$.subscribe({
      error: err => console.error('websocket connection error:', err),
      complete: () => console.log('websocket connection closed')
    });
  }

  // 暴露给组件的消息订阅入口
  public onMessage(): Observable<any> {
    return this.messageStream$;
  }
}

组件代码修正 #

import { Component, OnInit } from '@angular/core';
import { WebsocketService } from './websocket.service';

@Component({
  selector: 'app-demo',
  templateUrl: './demo.component.html'
})
export class DemoComponent implements OnInit {
  constructor(private wsService: WebsocketService) {}

  ngOnInit(): void {
    // 订阅消息流，仅处理业务逻辑
    this.wsService.onMessage().subscribe({
      next: (msg: any) => {
        console.log('component received message:', JSON.stringify(msg));
        this.handleMessage(msg);
      }
    });

    // 初始化WebSocket连接
    this.wsService.connect();
  }

  private handleMessage(msg: any): void {
    // 组件内部业务处理逻辑
  }
}

关键最佳实践点 #

• 流封装与隔离：通过RxJS操作符将错误、日志等非业务逻辑隔离在服务内部，组件只接收干净的消息流。
• 连接共享：使用share()操作符确保多组件订阅时复用同一个WebSocket连接，避免资源浪费。
• 生命周期合规：组件在ngOnInit中处理订阅和连接初始化，符合Angular组件生命周期规范，避免构造函数副作用。
• 错误边界：服务统一处理WebSocket错误，组件无需感知底层连接异常，降低耦合度。

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