RxJSを使用したリアクティブプログラミングの基本。 パート3.高階オブザーバブル

この記事では、あるスレッドで別のスレッドを処理する方法、それが必要な理由、および高次オブザーバブル（以下HOO）演算子がこれにどのように役立つかを見ていきます。



スレッドで作業する場合、別の作業の結果を値として1つのスレッドに転送する必要がある場合によく発生します。 たとえば、現在のスレッドでajaxリクエストを実行してそのレスポンスを処理したり、いくつかの並列リクエストを実行してプーリングを実装したりします。 約束などのメカニズムを使用してこのような問題を解決することに多くの人が慣れていると思います。 しかし、RxJSを使用してそれらを解決することは可能ですか？ もちろん、すべてがあなたが考えるよりもはるかに簡単です！



一連の記事「RxJSを使用したリアクティブプログラミングの基礎」：

• パート1.反応性とフロー
• パート2.演算子とパイプ

注 ：記事の理論的な部分を理解するために、以前の記事を読む必要はありません。観測可能な演算子、パイプが何であるかを知る必要があります。 実際の部分では、 2番目の記事の例を改良します 。

問題

次のタスクを想像してみましょう。サーバーにアクセスできるかどうかを毎秒調べる必要があります。 どうすれば解決できますか？



まず、タイマーメソッドを使用してストリームを作成します。

timer(0, 1000).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

タイマーメソッドは、原則としてintervalと非常に似ています 。 しかし、それとは異なり、スレッド開始タイムアウトを設定できます。これは、最初のパラメーターによって送信されます。 2番目のパラメーターは、新しい値が生成される間隔を示します。 2番目のパラメーターが指定されていない場合、タイマーは1つの値のみを生成し、ストリームを終了します。



あなたと私にはサーバーがないので、サーバーへのリクエストをエミュレートする関数を書くことをお勧めします。

 const makeRequest = () => { return timer(1000).pipe( mapTo('success') ) }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このメソッドは何をしますか？ タイマーメソッドを使用して作成されたストリームを返します。このメソッドは、1秒が経過して終了した後に値を発行します。 タイマーメソッドは数値のみを生成するため、mapTo演算子を使用して文字列「success」に置き換えます。



これは、makeRequestメソッドによって作成されたストリームの外観です。







ストリーム内でmakeRequestメソッドを呼び出すか、この責任をオブザーバーに割り当てるかを選択できます。



この場合、RxJSのすべての可能性を演算子で使用し、オブザーバーを不要な義務から解放できるため、最初のアプローチが望ましいです。 タイマーメソッドを使用して、間隔ごとにクエリを実行します。

 timer(0, 1000).pipe( map(() => makeRequest()) ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このようなコードを実行すると、console.logで、テキスト「success」を含むメッセージではなく、Observable型のオブジェクトを受信して​​いることがわかります。







マップではストリームを返すため、答えは非常に期待されています。 ストリームを機能させるには、サブスクライブする必要があります。 さて、 それをしない方法を見てみましょう：

 timer(0, 1000).pipe( map(() => makeRequest()) ).subscribe({ next: observable => observable.subscribe({ next: console.log }); });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

上記の例の問題は、サブスクリプションでサブスクリプションを取得することです。 しかし、チェーンで複数のリクエストを行いたい場合はどうでしょうか？ または、ある時点で内部のフローのサブスクリプションを解除する必要がある場合はどうなりますか？ この場合、コードはますます「麺」に似てきます。 この問題を解決するために、RxJSにはHOOと呼ばれる特別な演算子があります。

フー

HOOは、値としてストリームを受け入れる特別なタイプのステートメントです。 そのような演算子の1つにmergeAllメソッドがあります。



ストリームがmergeAllに到着すると、サブスクライブします。 オペレーターがサブスクライブしたストリームは、内部と呼ばれます。 オペレーターが値として他のフローを受け取るストリームは、外部と呼ばれます。



内部スレッドが値を生成すると、mergeAllはその値を外部スレッドにプッシュします。 したがって、手動でサブスクライブする必要がなくなります。 外部ストリームの購読を解除すると、mergeAllは内部ストリームの購読を自動的に解除します。



mergeAllを使用して例を書き換える方法を見てみましょう。

 timer(0, 1000).pipe( map(() => makeRequest()) mergeAll() ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

上記の例では、外部ストリームはタイマーステートメントによって作成されました。 また、マップ演算子で作成されるフローは内部です。 作成された各スレッドは、mergeAllステートメントに分類されます。







組み合わせマップ+ mergeAllは非常に頻繁に使用されるため、RxJSにはmergeMapメソッドがあります。

 timer(0, 1000).pipe( mergeMap(() => makeRequest()) ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

外部スレッドが値を生成すると、mergeMapオペレーターは渡されたコールバック関数を呼び出して、新しいスレッドを生成します。 次に、mergeMapは生成されたストリームにサブスクライブします。







mergeAll / mergeMap演算子の特性は、別のストリームがそれに到達すると、サブスクライブすることです。 したがって、外部ストリームでは、一度に複数の内部ストリームから値を取得できます。 次の例を見てみましょう。

  timer(0, 1000)
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これは、mergeMap演算子を使用しない場合の外部ストリームの外観です。







したがって、mergeMapの場合：

 timer(0, 1000).pipe( mergeMap(() => interval(1000)) )
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

毎秒、新しい内部スレッドが作成され、mergeMapがサブスクライブします。 したがって、多数の内部スレッドが同時に動作し、その値は外部になります。











注 ：mergeMapには注意してください 。新しい内部スレッドは、外部スレッドからサブスクライブを解除するまで機能します。 上記の例では、内部スレッドの数は毎秒増加しています。最終的には、コンピューターが負荷に対応できないほど多くのスレッドが存在する可能性があります。

concatAll / concatMap

mergeMapメソッドは、内部スレッドの実行順序を気にしない場合に最適ですが、必要な場合はどうでしょうか？ 前のサーバーからの応答を受信した後にのみ、次のサーバーリクエストを実行したいとしますか？



そのような目的には、HOO演算子concatAll / concatMapが適しています。 内部スレッドにサブスクライブしたこのオペレーターは、終了するまで待機し、次のスレッドにのみサブスクライブします。



あるスレッドの実行中に新しいスレッドが到達すると、前のスレッドが完了するまでそのスレッドがキューに置かれます。

 // ,  1     const firstInnerObservable = timer(1000).pipe( mapTo(1) ); // ,  2     const secondInnerObservable = timer(500).pipe( mapTo(2) ); of( firstInnerObservable, secondInnerObservable ).pipe( concatAll() ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

上記の例では、タイマーメソッドを使用して2つのスレッドを作成します。 わかりやすくするために、mapTo演算子を使用して異なる値を表示しました。 最初のスレッドは1を生成し、2番目のスレッドは2を生成します。ofメソッドを使用して外部スレッドが作成され、2つの監視可能な入力を入力として受け取ります。



concatAllステートメントは最初にfirstInnerObservableを受け取り、それにサブスクライブし、それが完了するのを待ちます。これは、secondInnerObservableに最初にサブスクライブした後のみです。 外部ストリームは次のようになります。







concatAllをmergeAllに置き換えると、ストリームは次のようになります。

 of( firstInnerObservable, secondInnerObservable ).pipe( mergeAll() ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

switchAll / switchMap

この演算子は、新しいストリームを受信するとすぐに以前のストリームからサブスクライブを解除し、新しいストリームをサブスクライブするという点で、以前のものとは異なります。



上記の例を使用し、concatAllをswitchAllに置き換えて、外部フローの動作を確認します。

 of( firstInnerObservable, secondInnerObservable ).pipe( switchAll() ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

2番目の内部ストリームの値のみが外部ストリームに入りました。 これは、switchMapが2番目のスレッドを受け取ったときに、最初からswitchMapがサブスクライブされていないためです。



これはいつ必要ですか？ たとえば、データ検索を実装する場合。 サーバーからの回答がまだ到着しておらず、既に新しいリクエストを送信している場合、前のリクエストを待つ必要はありません。

排気/排気マップ

exhaustはswitchAllステートメントの正反対であり、その動作はconcatAllに似ています。 ストリームにサブスクライブするこのメソッドは、ストリームが完了するまで待機します。 新しいストリームが届くと、それは単に破棄されます。

 of( firstInnerObservable, secondInnerObservable ).pipe( exhaust() ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

上記の例では、その時点でオペレーターが最初のスレッドの完了を待っていて、単に2番目のスレッドをドロップしたため、デュースは取得しませんでした。



多くの人が疑問を持っていると思います。いつそのような行動が必要になるのでしょうか？ 良い例がログインフォームです。 現在の要求が完了するまで、複数の要求をサーバーに送信することは意味がありません。

私たちはアプリケーションを完成させています

2番目の記事の 例を思い出します 。 その中で、GitHubで検索を実装し、mergeMap演算子を使用してサーバーにリクエストを送信しました。 これで、この演算子の機能がわかりました。私たちの場合、本当に適していますか？

 fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(val => val.length > 2), distinctUntilChanged(), mergeMap(value => { return from(getUsersRepsFromAPI(value)).pipe( catchError(err => of([])) ) }) ).subscribe({ next: reps => recordRepsToList(reps) })
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

GitHubサーバーが非常に過負荷になり、応答の処理に時間がかかると仮定しましょう。 この場合、何が問題になる可能性がありますか？



ユーザーがデータを入力し、回答を待たずに新しいデータを入力したとします。 この場合、2番目の要求をサーバーに送信します。 ただし、最初の要求に対する回答がより早く来ることを保証する人はいません。



mergeMapオペレーターは内部フローを処理する順序を気にしないため、最初の要求が2番目の要求よりも後に実行される場合、実際のデータを消去します。 したがって、mergeMapメソッドをswitchMapに置き換えることを提案します。

 fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(val => val.length > 2), distinctUntilChanged(), switchMap(value => { return from(getUsersRepsFromAPI(value)).pipe( catchError(err => of([])) ) }) ).subscribe({ next: reps => recordRepsToList(reps) })
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ユーザーが新しいデータを入力すると、switchMapは以前のストリームからサブスクライブを解除し、新しいストリームをサブスクライブします。



サーバーが応答するまで、httpリクエストがハングし続けることに注意してください。 しかし、内部ストリームからサブスクライブされていないため、答えは外部ストリームに分類されません。



注 ：Angularを使用し、HttpClientを使用してhttpを使用する場合、リクエスト自体をキャンセルすることはできません。 HttpClientは、登録解除時にこれを行うことができます。

httpをキャンセル

フェッチAPIには、 AbortControllerを使用してhttpリクエストをキャンセルする機能があります 。 switchMap演算子と組み合わせると、この機能はユーザートラフィックを節約します。



この例を少し書き換えましょう。 そして、フェッチ呼び出しをobservableでラップするメソッドを作成します。

 const createCancellableRequest = (url) => { //      const controller = new AbortController(); const signal = controller.signal; return new Observable(observer => { fetch(url, { signal }) .then(response => { if (response.ok) { return response.json(); } throw new Error(''); }) //     .then(result => observer.next(result)) //   .then(() => observer.complete()) //   ,     .catch(error => observer.error(error)); // ,    return () => { //   controller.abort(); }; }); };
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

また、getUsersRepsFromApiメソッドを変更します。

 const getUsersRepsFromAPI = (username) => { const url = `https://api.github.com/users/${ username }/repos`; return createCancellableRequest(url); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

現在、このメソッドはpromiseではなく、observableを返します。 したがって、switchMapのfromラッパーを削除します。

 switchMap(value => { return getUsersRepsFromAPI(value).pipe( catchError(err => of([]) ) )
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

注 ：RxJSバージョン6.5では、 fromFetchステートメントが追加され、それ自体が内部で abortメソッドを呼び出すため、独自の「バイク」を記述する必要がなくなりました。



それだけです！ すべてのサンプルコードはここにあります 。

おわりに

今日は、HOOとは何か、このカテゴリの非常に便利な演算子をいくつか見てきました。 もちろん、これらはすべてとは程遠いものでした。 より詳細な情報については、RxJSのドキュメントをご覧になることをお勧めします 。



次の記事では、ホットとコールドのオブザーバブルの違いを検討する予定です。



最後に、HOOがあるため、サブスクリプションでサブスクリプションを使用しないでください！