RxJS：N値を生成するswitchMapを使用した背圧

Question

ページ分割を含むいくつかのRESTful検索エンドポイントと話しています。クエリは、ユーザが検索フィールドに入力することによってトリガされ、その結果、N個の結果ページに対応するN値のObservableが生成されます。

コードは多少、以下のようになります。今

function runQueries(queryObservable) { 
    return queryObservable 
      .debounceTime(500) 
      .distinctUntilChanged() 
      .switchMap(search); 
} 

function search(query) { 
    return Observable.create(observer => searchInto(observer, query, 0)); 
} 

function searchInto(observer, query, start) { 
    runQuery(query, start).subscribe(result => { 
    observer.next(result); 
    if (hasMorePages(result)) { 
     searchInto(observer, query, start + 1); 
    } else { 
     observer.complete(); 
    } 
    }); 
} 

、検索要求にはしばらく時間がかかることができ、ユーザーがクエリを変更した場合、私はすべてのページを取得する必要はありません。

検索で3ページが返され、1ページが読み込まれた後にユーザーがクエリを変更したとします。

USER: types query A 
CODE: loads page A1 
USER: types query B 
CODE: loads page B1 
CODE: loads page B2 
CODE: loads page B3 

switchMapは仕事の半分を取得します。私のようなものを見てみたいです。得られた観測値は、A1、B1、B2、B3という正しい順序を持ちます。すばらしいです。

しかし、私の再帰的な検索では依然としてすべてのクエリが実行されており、サーバー、ネットワークなどに不必要な負荷がかかります。switchMapは「古い」結果を破棄しますが、再帰関数が終わりまで働きなさい。言い換えれば、それは次のようになります：

USER: types query A 
CODE: loads page A1 -> returned by search observable 
USER: types query B 
CODE: loads page A2 -> discarded by search observable 
CODE: loads page B1 -> returned by search observable 
CODE: loads page B2 -> returned by search observable 
CODE: loads page A3 -> discarded by search observable 
CODE: loads page B3 -> returned by search observable 

"A"と "B"のシーケンスはランダムです（競合条件に従います）が、問題はありません。

私は間違っていますか？これに対する慣用的な解決策は何ですか？

Answer 1

switchMapだけで操作をキャンセルすることができます。 SubscriptionをObservable.createに返信していないため、機内操作をキャンセルすることはできません。

それは本当にこのケースであなたを助けていないObservable.createを使用して立っているように、私はあなたの代わりに再帰的な操作を実行するexpandオペレータを利用することを示唆している：

function runQueries(queryObservable) { 
    return queryObservable 
      .debounceTime(500) 
      .distinctUntilChanged() 
      .switchMap(search); 
} 

function search(query) { 
    //Kicks off the first query 
    return runQuery(query, 0) 
     //Uses the results of the first query to see if more queries should be made 
    .expand((result, idx) => 
     //Continues to execute more queries until `hasMorePages` is false 
     hasMorePages(result) ? 
     runQuery(query, idx + 1) : 
     Observable.empty()); 
} 

Answer 2

flatMapは、新しい「ストリーム」を開始するときにObservableからの退会に十分親切です。プロデューサーは、その作業を停止するobserver.isUnsubscribedとサブスクリプションのステータスを使用することができます：あなたが解約されObservableを返す場合

function searchInto(observer, query, start) { 
    runQuery(query, start).subscribe(result => { 
    observer.next(result); 
    if (hasMorePages(result) && !observer.isUnsubscribed) { 
     searchInto(observer, query, start + 1); 
    } else { 
     observer.complete(); 
    } 
    }); 
}