如何在不关闭底层WebSocket的前提下对WebSocketSubject使用firstValueFrom

方案1：加空订阅保活（零侵入改造成本最低） #

WebSocketSubject默认会在所有活跃订阅都退订时关闭底层连接，你只需要提前给实例加一个永久空订阅占住引用计数，firstValueFrom的临时订阅退订后就不会触发连接关闭：

import { WebSocketSubject } from 'rxjs/webSocket';
import { firstValueFrom, filter } from 'rxjs';

let websocket = new WebSocketSubject<any>(url);
// 新增空订阅保活，连接不会自动关闭
const keepAliveSub = websocket.subscribe();

// 原有逻辑完全不需要修改
let firstMessage = await firstValueFrom(websocket.pipe(filter(m => true)));

// 后续需要主动关闭连接时，再取消保活订阅即可
// keepAliveSub.unsubscribe()

方案2：用share操作符封装多播层（无额外维护成本） #

如果不想手动管理保活订阅，可以用RxJS7+的share操作符配置多播层，拦截退订请求不传递给底层WebSocketSubject，所有收发逻辑和原生WebSocketSubject用法完全一致，不需要区分两个实例：

import { WebSocketSubject } from 'rxjs/webSocket';
import { firstValueFrom, filter, share } from 'rxjs';

let websocket = new WebSocketSubject<any>(url).pipe(
  share({
    // 引用计数归零时不取消源订阅，不会关闭WebSocket
    resetOnRefCountZero: false
  })
);

// 正常调用firstValueFrom即可，不会触发连接关闭
let firstMessage = await firstValueFrom(websocket.pipe(filter(m => true)));
// 发送上行消息逻辑也不变，直接调用next即可
websocket.next({ type: 'ping' });

// 需要关闭连接时直接调用底层subject的complete方法即可
// websocket.source.complete()

原理解释 #

两个方案都是基于WebSocketSubject的默认行为设计：只有当所有直接订阅源Subject的订阅都退订时，才会触发底层连接关闭。方案1是用空订阅占住引用计数，方案2是用多播层把所有订阅都拦截在share内部，退订只会减少多播层的引用计数，不会传递给源WebSocketSubject，自然不会触发连接关闭。
两个方案都不需要重写firstValueFrom、不需要维护双实例、不需要用到冗余的multiplex能力，完全匹配需求。

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