RxJSを使用したリアクティブプログラミングの基本。 パート2.演算子とパイプ

前回の記事では、ストリームとは何か、そしてそれらが何を食べるのかを見ました。 新しいパートでは、RxJSがフローを作成するために提供するメソッド、オペレーター、パイプとは何か、そしてそれらを操作する方法について学びます。



RxJSには豊富なAPIがあります 。 ドキュメントには、100を超えるメソッドが説明されています。 それらについて少し知るために、単純なアプリケーションを作成し、実際にはリアクティブコードがどのように見えるかを確認します。 反応性のプリズムを通してそれらを見ると、かつては日常的で多くのコードを書く必要があった同じタスクがエレガントなソリューションを持っていることがわかります。 しかし、実践を始める前に、フローをグラフィカルに表現する方法を見て、フローを作成して処理する便利な方法を理解します。

スレッドのグラフィカルな表現

特定のフローがどのように動作するかを明確に示すために、リアクティブアプローチで採用されている表記法を使用します。 前の記事の例を思い出してください。

const observable = new Observable((observer) => { observer.next(1); observer.next(2); observer.complete(); });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そのグラフィカル表現は次のようになります。







通常、フローは直線で表されます。 ストリームが値を出力する場合、線として円として表示されます。 ディスプレイ内の直線は、ストリームを終了する信号です。 エラーを表示するには、記号「-」を使用します。

 const observable = new Observable((observer) => { observer.error(); });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ワンラインストリーム

私の練習では、自分のObservableインスタンスを直接作成する必要はほとんどありませんでした。 スレッドを作成するほとんどのメソッドは、すでにRxJSにあります。 値1および2を放出するストリームを作成するには、ofメソッドを使用します。

 const observable = of(1, 2);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ofメソッドは任意の数の引数を受け入れ、Observableの既製のインスタンスを返します。 購読後、受信した値を発行して完了します：







配列をストリームとして表現する場合は、fromメソッドを使用できます。 引数としてのfromメソッドは、反復可能なオブジェクト（配列、文字列など）またはプロミスを想定し、このオブジェクトをストリームに投影します。 文字列から取得したストリームは次のようになります。

 const observable = from('abc');
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、あなたはストリームで約束を包むことができます：

 const promise = new Promise((resolve, reject) => { resolve(1); }); const observable = from(promise);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

注：多くの場合、スレッドはpromiseと比較されます。 実際、共通点は1つだけです。変化を広めるためのプッシュ戦略です 。 残りは完全に異なるエンティティです。 Promiseは複数の値を生成できません。 解決または拒否のみを実行できます。 2つの状態のみがあります。 ストリームは複数の値を送信でき、再利用できます。



最初の記事からの間隔の例を覚えていますか？ このストリームは、サブスクリプションの瞬間から秒単位で時間をカウントするタイマーです。

 const timer = new Observable(observer => { let counter = 0; const intervalId = setInterval(() => { observer.next(counter++); }, 1000); return () => { clearInterval(intervalId); } });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

同じことを1行で実装する方法は次のとおりです。

 const timer = interval(1000);
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして最後に、DOM要素のイベントのストリームを作成できるメソッド：

 const observable = fromEvent(domElementRef, 'keyup');
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

値として、このストリームはキーアップイベントオブジェクトを送受信します。

パイプとオペレーター

Pipeは、RxJSバージョン5.5で追加されたObservableクラスメソッドです。 そのおかげで、ストリームで受信した値を順次処理するための演算子のチェーンを構築できます。 パイプは、オペレーターを相互接続する単方向チャネルです。 演算子自体は、ストリームからの値を処理するRxJSで記述されている通常の関数です。



たとえば、値を変換してストリームにさらに渡すことができます。また、フィルターとして機能し、指定された条件を満たしていない場合は値をスキップしません。



動作中の演算子を見てみましょう。 マップ演算子を使用して、ストリームの各値に2を掛けます。

 of(1,2,3).pipe( map(value => value * 2) ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

マップ演算子を適用する前のストリームは次のとおりです。







mapステートメントの後：







フィルター演算子を使用してみましょう。 このステートメントは、Arrayクラスのフィルター関数と同じように機能します。 このメソッドは、条件を記述する関数を最初の引数として受け取ります。 ストリームの値が条件を満たしている場合、次のように渡されます。

 of(1, 2, 3).pipe( //     filter(value => value % 2 !== 0), map(value = value * 2) ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

そして、これは私たちのストリームのスキーム全体がどのように見えるかです：







フィルター後：







マップ後：







注： pipe！==サブスクライブ。 pipeメソッドはフローの動作を宣言しますが、サブスクライブしません。 subscribeメソッドを呼び出すまで、ストリームは機能し始めません。

アプリケーションを書く

パイプと演算子が何であるかがわかったので、練習を開始できます。 アプリケーションは、1つの簡単なタスクを実行します。入力された所有者のニックネーム別に、開いているgithubリポジトリのリストを表示します。



いくつかの要件があります。

• 入力に入力された文字列に含まれる文字が3文字未満の場合、API要求を実行しないでください。
• ユーザーが入力した各文字の要求を満たさないようにするには、APIにアクセスする前に700ミリ秒の遅延（デバウンス）を設定する必要があります。

リポジトリを検索するには、 github APIを使用します 。 stackblitzでサンプルを実行することをお勧めします 。 そこで、完成した実装をレイアウトしました。 リンクは記事の最後にあります。



htmlマークアップから始めましょう。 input要素とul要素について説明しましょう。

 <input type="text"> <ul></ul>
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、jsまたはtsファイルで、ブラウザーAPIを使用して現在の要素へのリンクを取得します。

 const input = document.querySelector('input'); const ul = document.querySelector('ul');
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

また、github APIへのリクエストを実行するメソッドも必要です。 以下はgetUsersRepsFromAPI関数のコードです。この関数はユーザーのニックネームを受け入れ、フェッチを使用してajaxリクエストを実行します。 次に、約束を返し、成功した応答を途中でjsonに変換します。

 const getUsersRepsFromAPI = (username) => { const url = `https://api.github.com/users/${ username }/repos`; return fetch(url) .then(response => { if(response.ok) { return response.json(); } throw new Error(''); }); }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

次に、リポジトリの名前をリストするメソッドを作成します。

 const recordRepsToList = (reps) => { for (let i = 0; i < reps.length; i++) { //    ,    if (!ul.children[i]) { const newEl = document.createElement('li'); ul.appendChild(newEl); } //      const li = ul.children[i]; li.innerHTML = reps[i].name; } //    while (ul.children.length > reps.length) { ul.removeChild(ul.lastChild); } }
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

準備が完了しました。 RxJSの実際の動作を見てみましょう。 入力のキーアップイベントをリッスンする必要があります。 まず、リアクティブアプローチではフローを操作することを理解する必要があります。 幸いなことに、RxJSはすでに同様のオプションを提供しています。 前述のfromEventメソッドを思い出してください。 私たちはそれを使用します：

 const keyUp = fromEvent(input, 'keyup'); keyUp.subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これで、イベントがストリームとして表示されます。 コンソールに表示されるものを見ると、KeyboardEventタイプのオブジェクトが表示されます。 ただし、ユーザーが入力した値が必要です。 これは、パイプメソッドとマップ演算子が役立つ場所です。

 fromEvent(input, 'keyup').pipe( map(event => event.target.value) ).subscribe({ next: console.log });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

要件の実装に進みます。 まず、入力した値に3文字以上が含まれている場合にクエリを実行します。 これを行うには、フィルター演算子を使用します。

 fromEvent(input, 'keyup').pipe( map(event => event.target.value), filter(value => value.length > 2) )
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最初の要件が整理されました。 2番目に進みます。 デバウンスを実装する必要があります。 RxJSにはdebounceTimeステートメントがあります。 最初の引数としてのこの演算子は、値が保持される間、渡す前にミリ秒数をとります。 この場合、新しい値ごとにタイマーがリセットされます。 したがって、出力では、700ミリ秒が経過した後、最後の値を取得します。

 fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(value => value.length > 2) )
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

debounceTimeを使用しない場合、ストリームは次のようになります。







そして、これは、このステートメントを通過した同じストリームがどのように見えるかです：







debounceTimeを使用すると、APIを使用する可能性が低くなり、トラフィックを節約してサーバーの負荷を軽減できます。



追加の最適化のために、別の演算子distinctUntilChangedを使用することをお勧めします。 この方法により、重複を防ぐことができます。 例でその作業を示すのが最善です：

 from('aaabccc').pipe( distinctUntilChanged() )
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

distinctUntilChangedなし：







distinctUntilChangedの場合：







debounceTimeステートメントの直後にこのステートメントを追加します。 したがって、何らかの理由で新しい値が前の値と一致する場合、APIにアクセスしません。 同様の状況は、ユーザーが新しい文字を入力してから再び消去したときにも発生します。 遅延を実装しているので、最後の値のみがストリームに含まれます。これは既に持っている答えです。

サーバーに行く

これで、リクエストとレスポンスの処理のロジックを説明できます。 私たちは約束をもってのみ働くことができますが。 したがって、getUsersRepsFromAPIメソッドを呼び出す別のマップ演算子について説明します。 オブザーバーでは、約束の処理ロジックを記述します。

 /*  !     RxJS    promise,      */ fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(val => val.length > 2), distinctUntilChanged(), map(value => getUsersRepsFromAPI(value)) ).subscribe({ next: promise => promise.then(reps => recordRepsToList(reps)) });
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

現時点では、必要なすべてを実装しています。 しかし、この例には1つの大きな欠点があります。エラー処理がありません。 私たちのオブザーバーは約束だけを受け取り、何かがうまくいかない可能性があることを知りません。



もちろん、次のメソッドでpromiseにcatchを掛けることができますが、このため、コードはますます「コールバック地獄」のように見え始めます。 突然もう1つのリクエストを実行する必要がある場合、コードの複雑さが増します。



注： RxJSコードでpromiseを使用することは、アンチパターンと見なされます。 Promiseには、観測可能と比較して多くの欠点があります。 元に戻すことはできず、再利用することもできません。 選択肢がある場合は、observableを選択します。 ObservableクラスのtoPromiseメソッドについても同じことが言えます。 このメソッドは、ストリームを処理できないライブラリとの互換性のために実装されました。



fromメソッドを使用してプロミスをストリームに投影できますが、このメソッドにはsubscribeメソッドへの追加の呼び出しが含まれており、コードの成長と複雑さにもつながります。



この問題は、mergeMap演算子を使用して解決できます。

 fromEvent(input, 'keyup').pipe( debounceTime(700), map(event => event.target.value), filter(val => val.length > 2), distinctUntilChanged(), mergeMap(value => from(getUsersRepsFromAPI(value))) ).subscribe({ next: reps => recordRepsToList(reps), error: console.log })
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

ここで、Promise処理ロジックを記述する必要はありません。 fromメソッドはpromiseストリームを作成し、mergeMapオペレーターがそれを処理しました。 約束が正常に履行されると、次のメソッドが呼び出され、オブザーバーは完成したオブジェクトを受け取ります。 エラーが発生した場合、エラーメソッドが呼び出され、オブザーバーはコンソールにエラーを出力します。



mergeMap演算子は、以前に使用した演算子とは少し異なり、次の記事で説明するいわゆるHigher Order Observablesに属します。 しかし、先を見て、私はmergeMapメソッド自体がストリームをサブスクライブすると言います。

エラー処理

スレッドがエラーを受信した場合、スレッドは終了します。 また、エラーの後にアプリケーションと対話しようとしても、ストリームが完了しているため、反応はありません。



ここで、catchError演算子が役立ちます。 catchErrorは、ストリームでエラーが発生した場合にのみ発生します。 インターセプトして処理し、通常の値をストリームに戻すことができますが、完了には至りません。

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

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

catchErrorでエラーをキャッチし、代わりに空の配列を持つストリームを返します。 ここで、エラーが発生した場合、リポジトリのリストをクリアします。 しかし、その後、フローは再び終了します。



問題は、catchErrorが元のストリームを新しいストリームに置き換えることです。 そして、私たちのオブザーバーは彼だけに耳を傾けます。 ofストリームが空の配列を発行すると、completeメソッドが呼び出されます。



元のスレッドを置き換えないために、mergeMap演算子内からfromスレッドでcatchError演算子を呼び出します。

 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), error: console.log })
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

したがって、元のストリームは何にも気付きません。 エラーの代わりに、空の配列を取得します。

おわりに

ようやく練習に取り掛かり、パイプとオペレーターの目的を確認しました。 RxJSが提供するリッチAPIを使用してコードを短縮する方法を検討しました。 もちろん、アプリケーションは終了していません。次のパートでは、1つのスレッドで別のスレッドを処理する方法と、アプリケーションのトラフィックとリソースをさらに節約するためにhttpリクエストをキャンセルする方法を分析します。 そしてその違いを見ることができるように、RxJSを使用しないで例をレイアウトしました 。 ここでそれを見ることができます 。 このリンクには、現在のアプリケーションの完全なコードがあります。 回路を生成するために、 RxJSビジュアライザーを使用しました 。



この記事が、RxJSがどのように機能するかをよりよく理解するのに役立つことを願っています。 勉強に成功したいです！