浅淡 RxJS WebSocket

引言

中后台仪表盘是一个很是复杂，特别是当须要全面屏运用时，数据的实时性需求很是高。WebSocket 无论在什么环境中使用其实都是很是简单，各现代浏览器实现标准都很统一，并且接口也足够简单。

即使是在 Angular 也是如此，只须要简单几行代码就能使用 WebSocket。

const ws = new WebSocket('wss://echo.websocket.org');
ws.onmessage = (e) => {
  console.log('message', e);
}

若须要向服务端发送消息，则：

ws.send(`content`);

在 Angular 里绝大多数的人都会根据上述代码进一步拓展，好比统一消息解析、错误处理、多路复用等，并最终将其封装成一个服务类。

事实上，RxJS 也包裹了一个 WebSocket Subject，位于 rxjs/websocket。

如何使用

假如将上面的示例使用 RxJS 来写，则：

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

const ws = webSocket('wss://echo.websocket.org');

ws.subscribe(res => {
  console.log('message', res);
});

ws.next(`content`);

webSocket 是一个工厂函数，所生产出来的 WebSocketSubject 对象可被屡次订阅，若未订阅或取消最后一个订阅时都会致使 WebSocket 链接中断，当再一次订阅时会从新自动链接。

WebSocketSubjectConfig

webSocket 除了接收字符串（WebSocket服务远程地址）外，还容许指定更复杂的配置项。

默认状况下，消息是使用 JSON.parse 和 JSON.stringify 对消息格式序列化和反序列化操做，因此无论消息发送或接收都以 JSON 为准，可经过 serializer、deserializer 属性来改变。

若须要关心 WebSocket 何时开始或结束（closeObserver），则：

const open$ = new Subject();
const ws = webSocket({
  url: 'wss://echo.websocket.org',
  openObserver: open$
});
// 订阅打开事件
open$.subscribe(() => {});

消息

WebSocketSubject 也是 Subject 的变体之一，所以订阅它表示接收消息，反之则利用 next、complete、error 来维护消息的推送。

• 使用 next 来发送消息
• 使用 complete 会尝试检测是否最后一个订阅，如果将会关闭链接
• 使用 error 至关于原始 close 方法且必须提供 { code: number, reason?: string} 参数，注意 code 务必遵照取值范围

可被重放

调用 next 发送消息时若 WebSocket 链接中断（例如：没人订阅时），消息会被缓存当下一次从新链接之后会按顺序发送。这对于异步世界里很是方便，咱们只须要确保 Angular 启动前初始化好 WebSocket 无论何时订阅接收消息，均可以随时发送也无须等待。

事实上这一点是 RxJS WebSocket 默认状况下是经过 webSocket 所生产的 WebSocketSubject 其本质上是 ReplaySubject 的“重放”能力。固然你能够经过 webSocket 的第二个参数改变这种行为。

多路复用

通常来讲咱们不太可能只会一个 Web Socket 服务完成全部的事，然而也不太可能针对每个业务实例建立一个 webSocket。每每咱们会增长一层网关并将这些业务 WebSocket 进行汇总，对于前端始终只须要一个链接，这就是多路复用存在的意义。

而核心是必需要让后端知道，何时发送什么消息给什么样的服务。

首先必须先使用 multiplex 方法来建立 Observable 以便订阅某一路消息，它有三个参数来帮助咱们区分消息：

• subMsg 告知正在订阅哪一路消息
• unsubMsg 告知取消订阅哪一路消息
• messageFilter 过滤消息，使订阅者只接收哪一路消息

const ws = webSocket('wss://echo.websocket.org');
const user$ = this.ws.multiplex(
    () => ({ type: 'subscribe', tag: 'user' }),
    () => ({ type: 'unsubscribe', tag: 'user' }),
    message => message.type === 'user'
);
user$.subscribe(message => console.log(message));

const todo$ = this.ws.multiplex(
    () => ({ type: 'subscribe', tag: 'todo' }),
    () => ({ type: 'unsubscribe', tag: 'todo' }),
    message => message.type === 'todo'
);
todo$.subscribe(message => console.log(message));

user$ 流和 todo$ 流他们共用一个 WebSocket 链接，这即是多路复用。

虽然订阅是经过 multiplex 建立的，而后消息的推送依然仍是须要使用 ws.next()。

总结

这本来是对内部一个简单培训，然而我发现居然极少人会讨论 RxJS 里面 Web Socket 的实现。

其实一直有想着要给 ng-alain 内置 WebSocket，只是就封装角度来说彻底没有价值，由于已经足够优雅。