淺淡 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，只是就封裝角度來說徹底沒有價值，由於已經足夠優雅。