RxJS5 WebSocketSubject - メッセージをフィルタリングして完成させるには？

Question

私はRxJS 5とのWebSocket接続を設定する正しい方法についてのガイダンスを探しています。JSON-RPC 2.0を使用するWebSocketに接続しています。私はWSに要求を送信し、サーバからの関連する応答のObservableを返す関数を実行できるようにしたい。私はそうのように私の最初のWebSocketSubjectを設定

：この観察できることから

const ws = Rx.Observable.webSocket("<URL>")

、私はws.next(myRequest)を使用して要求を送信することができた、と私は応答がws`て戻って来るのを見ることができました観察可能である。

私は正しいレスポンスにwsレスポンスをフィルタリングして完了する関数の作成に苦労しました。これらはソースの件名を完成させるように見え、将来のすべてのリクエストを停止します。私は次のことを試してみました

function makeRequest(msg) { 
    // 1. send the message 
    // 2. return an Observable of the response from the message, and complete 
} 

：

function makeRequest(msg) { 
    const id = msg.id; 
    ws.next(msg); 
    return ws 
     .filter(f => f.id === id) 
     .take(1); 
} 

私はしかし、唯一の最初の要求が動作することを行うと

私の意図した出力のようなものです。それ以降のリクエストはうまくいかず、take(1)で完了していると思いますか？

このような状況に適したアーキテクチャについて考えていますか？

Answer 1

登録者がいない場合は、登録解除の際にWebSocketを閉じるバグか慎重な意思決定があるようです。興味がある場合はsourceとなります。

本来、加入者が常にあることを保証する必要があります。それ以外の場合はWebSocketが閉鎖されます。あなたは2つの方法でこれを行うことができます。

ルートAはより意味のある方法です。本質的には、SubjectのObservable部分の公開バージョンを作成し、より詳細な制御が可能です。

const ws = Rx.Observable.webSocket("<URL>"); 
const ws$ = ws.publish(); 

//When ready to start receiving messages 
const totem = ws$.connect(); 

function makeRequest(msg) { 
    const { id } = msg; 
    ws.next(msg); 
    return ws$.first(f => f.id === id) 
} 

//When finished 
totem.unsubscribe(); 

ルートBは、単純にソケットを保持しているトークンのサブスクリプションを作成することですが、あなたのアプリケーションの実際のライフサイクルに応じて、あなたはそれを常に取得することを確認するために、closingイベントのいくつかの並べ替えに接続するとよいでしょう閉鎖した。すなわち

const ws = Rx.Observable.webSocket("<URL>"); 

const totem = ws.subscribe(); 

//Later when closing: 
totem.unsubscribe(); 

ご覧のとおり、どちらの方法もどちらもサブスクリプションを作成するので、どちらのアプローチも同様です。 Bの主な欠点は、空のサブスクリプションを作成して、すべてのイベントをポンピングして投棄することだけです。 Bの利点は、Subjectと同じ変数を使用して放出とサブスクリプションを参照することができますが、サブスクリプションにはws$を使用していることに注意する必要があります。

あなたはSubject作成機能を使用してルートAを絞り込むことができ本当にそう傾斜した場合：

const safeWS = Rx.Subject.create(ws, ws$); 

上記は、あなたが同じ変数を使用できるようになる、あなたはまだws$をシャットダウンするための責任を負うことになるとあなたがそれを完了したときに、WebSocket、transitively。