Rxjsを使いこなせるようになりたい

みんな大嫌い 非同期処理
できることなら、やりたくない。

今回Angularフレームワークの依存ライブラリ Rxjs について
書いていきます。
少しでも理解につながればよいかなと思います。

# 1.Rxjsとは
RxJS は Reactive Extensions for JavaScript の略称です。Observables というアーキテクチャを用いたリアクティブ・プログラミング用のライブラリであり、非同期処理 ( クリックなどイベント駆動の処理も含まれます ) を簡潔かつ可読性高くコーディング出来ることを主な目的としています。

引用元：[RxJS を学ぼう #1 - これからはじめる人のための導入編](https://tech.recruit-mp.co.jp/front-end/rxjs-intro/)

Reactive Extensions ( Rx )というライブラリがさまざまな言語向けに移植されています。
極端に言うとJavaScriptでのイベント非同期処理をラップし、扱いやすくするためのライブラリです。

#2.Observableとsubscribe
Rxjsを使うに当たってObservableとsubscribeは必須になります。

### Observable
Observableを実装する際は、データを流す(next)ことと、受け取る（subscribe）ことが必要になります。

川の流れをイメージすると良いかもしれません。
![image.png](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2280858/44f2bfa2-794a-a103-031e-1396081e3cf2.png)
野球ボールというデータ(next)を生成し
サッカーボールに加工(pipe)して
最終的に桃として受け取ります(subscribe)
例えがひどいですが、このような処理をします。


pipeはデータ(ストリーム)を流す中で加工を行う処理なのですが
後で説明します。

### Subscribe
受け取る（subscribe）処理になります。
その為Observableはデータを生成したら受け取る（subscribe）処理が必要になります。

以下例です
Angularの主要言語はTypeScriptになります。

```src\app\app.component.ts
import { Component, OnInit } from '@angular/core';
import { Observable } from 'rxjs';

@Component({
  selector: 'ac-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css'],
})
export class AppComponent implements OnInit {
  public testObservable: Observable<string> = new Observable((observer) => {
    observer.next('あいうえお');
    setTimeout(() => {
      observer.next('かきくけこ');
    }, 1000);
  });

  constructor() {}

  ngOnInit() {
    this.testObservable.subscribe((data: string) => {
      console.log(data);
      setTimeout(() => {
        console.log('さしすせそ');
      }, 1000);
    });
    setTimeout(() => {
      console.log('たちつてと');
    }, 500);
  }
}

```

実行結果

```
あいうえお
たちつてと
さしすせそ
かきくけこ
さしすせそ
```

ngOnInit()はangularで用意されているライフサイクルメソッドです。
今回はコンストラクタと同じ感覚で良いです。

1.textObservable()がsubscribe()を実行し、イベントが発火します。
observer.next('あいうえお');により、ストリームが発生します。
'あいうえお'を受け取り、console.logで表示します。

2.'たちつてと'は呼び出されてから0.5秒後に発火します。
setTimeout(() => {console.log('さしすせそ');}, 1000);を待ってから実行されてない為
この時点で非同期処理が行われていることがわかります。

3.'さしすせそ'は'あいうえお'を受け取った後1秒後に実行されます。

4.'かきくけこ'はobserver.next('あいうえお');を実行した後1秒後に実行されます。

5.最後の'さしすせそ'は'かきくけこ'を受け取っている為実施されます。

上記からわかる点は
・非同期処理であること。
・Observable型はストリームを流す(next)度に実行される。

ことになります。
ちなみにsubscribe()がない場合、next()を実施しても処理を行いません。

#3.Subject

上記のコードではObservableクラスのインスタンスを作成したタイミングでしかデータを流すことができず、クリックなどのイベントをトリガーとしてデータを処理したいような場合には向いていません。

そんな時に使用するのがSubjectクラスです。
SubjectクラスのインスタンスはObservableとobserverの２つの役割を同時に担うことができ、任意のタイミングでデータを流すことができます。

以下例です。

```src\app\app.component.ts
import { Component, OnInit } from '@angular/core';
import { Subject } from 'rxjs';

@Component({
  selector: 'ac-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css'],
})
export class AppComponent  implements OnInit {

  public testSubject: Subject<number> = new Subject();
  public testState = this.testSubject.asObservable();
  public count = 0;

  constructor() {
  }

  ngOnInit() {
    this.testState.subscribe((data: number) => {
      console.log(data);
    });
  }

  clickNext() {
    this.count++;
    this.testSubject.next(this.count);
  }

}
```

```src\app\app.component.html
<div class="text-center">
  <button class="btn text-center" (click)="clickNext()">Next</button>
</div>
```

testSubjectでSubjectのインスタンスを格納し、testStateにSubjectのObservableを格納します。
Subjectのインスタンスは前出のobserverと同じ振る舞いをするため、clickNext()が発火するたびにloginStateへデータを流していきます。

監視者って呼ばれ方もします。要は見張りです。
testStateっていう監視者を作って呼び出されるたびに非同期で処理しています。

![動作.gif](https://qiita-image-store.s3.ap-northeast-1.amazonaws.com/0/2280858/2e8afd9e-b3c0-896e-b4d2-449c0a7859a8.gif)

クリックするたびにデータが更新されることが確認できました。

#4.Pipe
最後にpipeです。
これはデータ(ストリーム)を流す中で加工処理をする際に使います。
先ほどのクリック処理に少し処理を加えてみます。

```src\app\app.component.ts
import { Component, OnInit } from '@angular/core';
import { Subject } from 'rxjs';
import { map } from 'rxjs/operators'; //追加

@Component({
  selector: 'ac-root',
  templateUrl: './app.component.html',
  styleUrls: ['./app.component.css'],
})
export class AppComponent implements OnInit {
  public testSubject: Subject<number> = new Subject();
  public testState = this.testSubject.asObservable();
  public count = 0;

  constructor() {}

  ngOnInit() {
    this.testState
    .pipe( //追加
      map((x) => x * 2) //追加
      ) //追加
    .subscribe((data: number) => {
      console.log(data);
    });
  }

  clickNext() {
    this.count++;
    this.testSubject.next(this.count);
  }
}

}
```

結果としては先ほどの実行結果を2倍にして表示されます。 2.4.6.8.10...
今回は簡単な処理で例をあげましたが
例えば非同期処理から非同期処理を呼び出したり、
複数の非同期処理をマージしたりなど
いろんな用途があります。

### 最後に
なんとなくお分かりいただけたでしょうか。

非同期処理はバグを生み出す代表格であり、実装がかなり大変です。
自分もこれを理解するまでかなりの時間がかかりました。
フロントをやっていく中で非同期処理は付いてくるので
抑えておくと後々楽になるかもしれませんね。

### 参考
[RxJSの基礎中の基礎](https://qiita.com/agajo/items/7942743a0130f7a0f30b)
[RxJS Observable まとめ](https://qiita.com/chibi929/items/4110ab0c777270f9b209)