RxJS6便利なよく使うObservableの使い方まとめ

2018年6月18日

この記事は公開されてから1年以上経過しています。情報が古い可能性がありますので、ご注意ください。

RxJXがv6へアップデートされ
オペレータ周りやインポートパスの変更が入ったため
あらためてRxJS6の便利なObservableの使い方をまとめました。

とりあえず動かすための設定から
紹介させていただきたいと思います。

環境

RxJS 6.2.0
typescript 2.9.1
parcel 1.8.1
Yarn 1.6.0

インストール

インストールはYarnで行いました。
RxJSをインストールします。

terminal$ yarn add rxjs

typescriptとRxJSの型ファイルと
動作環境用にparcelをインストールします。

terminal$ yarn add -D typescript @types/rx parcel-bundler

package.jsonへ下記を追加します。

package.json  "scripts": {
    "dev": "parcel index.html"
  },

index.htmlを追加します。

index.html<!DOCTYPE html>
<html lang="en">
<head>
  <meta charset="UTF-8">
  <meta name="viewport" content="width=device-width, initial-scale=1.0">
  <meta http-equiv="X-UA-Compatible" content="ie=edge">
  <title>Document</title>
</head>
<body>
    <ul id="itemLists"></ul>
    <script src="./src/app.ts"></script>
</body>
</html>

src/app.tsを追加します。

src/app.tsimport {
  of,
  from,
  fromEvent,
  fromEventPattern,
  interval,
  merge,
  concat,
  defer,
  empty,
  throwError,
  combineLatest,
} from 'rxjs';

これで動かす準備は整いました。早速Observableの使い方を見ていきましょう。

Observable

of

カンマ区切りの値を1つずつ流すObservableを生成します。

src/app.ts/*
* of
* カンマ区切りの値を1つずつ流すObservableを生成します。
* of<T>(...args: Array<T | SchedulerLike>): Observable<T>
*/
of(1, 2, 3).subscribe(
  (x: number): void => console.log(x)
);
// 結果
// 10
// 20
// 30

from

配列値やPromiseを流すObservableを生成します。

src/app.ts/*
* from
* 配列値やPromiseを流すObservableを生成します。
* from<T>(input: ObservableInput<T>, scheduler?: SchedulerLike): Observable<T>
*/
/* 配列*/
from([10, 20, 30]).subscribe(
  (x: number): void => console.log(x)
);
// 結果
// 10
// 20
// 30

/* 非同期*/
const request = 'https://api.github.com/search/repositories?q=tetris+language:assembly&sort=stars&order=desc';
const getSearchGithubList = (): Promise<object> =>
  fetch(request).then((respons: Response): Promise<object> => respons.json());

from(getSearchGithubList()).subscribe(
  (respons: object): void => console.log(respons)
);
// 結果
// {...}

fromEvent

Formのイベントを流すObservableを生成します。

src/app.ts
/*
* fromEvent
* Formのイベントを流すObservableを生成します。
* fromEvent<T>(target: EventTargetLike, eventName: string, options?: EventListenerOptions | ((...args: any[]) => T), resultSelector?: ((...args: any[]) => T)): Observable<T>
*/

const input = document.createElement('input');
document.body.appendChild(input);
fromEvent(input, 'change').subscribe(
  (event: Event): void => console.log(event);
);
// 結果
// テキスト入力すると
// Event {isTrusted: true, type: "change", target: input, currentTarget: input, eventPhase: 2, …}

fromEventPattern

Formのイベントを流すObservableを生成します。

src/app.ts/*
* fromEventPattern
* Formのイベントを流すObservableを生成します。
* fromEventPattern<T>(addHandler: (handler: Function) => any, removeHandler?: (handler: Function, signal?: any) => void, resultSelector?: (...args: any[]) => T): Observable<T | T[]>
*/

const button = document.createElement('button')
button.innerText = 'button';
document.body.appendChild(button);

const addClickHandler = (handler: () => void): void => {
  button.addEventListener('click', handler);
}

const removeClickHandler = (handler: () => void): void => {
  button.removeEventListener('click', handler);
}

fromEventPattern(addClickHandler, removeClickHandler).subscribe(
  (event: Event): void => console.log(event)
);

interval

一定時間ごとにObservableを生成します。

src/app.ts/*
* interval
* 一定時間ごとにObservableを生成します。
* interval(period: 0 = 0, scheduler: SchedulerLike = async): Observable<number>
*/

interval(1000).subscribe(
  (x: number): void => console.log(x)
);

// 結果
// 0
// 1
// 2

merge

複数のObservableをマージしたObservableを生成します。

src/app.ts/*
* marge
* 複数のObservableをマージしたObservableを生成します。
* merge<T, R>(...observables: Array<ObservableInput<any> | SchedulerLike | number>): Observable<R>
*/

merge(of('a', 'b', 'c'), of('d', 'e', 'f')).subscribe(
  (x: string): void => console.log(x)
);
// 結果
// a
// b
// c
// d
// e
// f

concat

複数のObservableを統合したObservableを生成します。

src/app.ts/*
* concat
* 複数のObservableを統合したObservableを生成します。
* concat<T, R>(...observables: Array<ObservableInput<any> | SchedulerLike>): Observable<R>
*/
concat(of('a', 'b', 'c'), of('d', 'e', 'f')).subscribe(
  (x: string): void => console.log(x)
);
// 結果
// a
// b
// c
// d
// e
// f

defer

Observerが購読する時点で動的にObservableを生成します。

src/app.ts/*
* defer
* Observerが購読する時点で動的にObservableを生成します。
* defer<T>(observableFactory: () => SubscribableOrPromise<T> | void): Observable<T>
*/

const button2 = document.createElement('button')
button2.innerText = 'button2';
document.body.appendChild(button2);

defer(() => {
  if (Math.random() > 0.5) {
    return fromEvent(button2, 'click');
  } else {
    return interval(1000);
  }
}).subscribe((x: number): void => console.log(x));
// 結果
// 0
// 1
// 2...
// or
// MouseEvent {isTrusted: true, screenX: 694, screenY: -1289, clientX: 204, clientY: 31, …}

empty

Observableの終了を返します。

src/app.ts/*
* empty
* Observableの終了を返します。
* empty(scheduler?: SchedulerLike)
*/
empty().subscribe(
  (x: any): void => console.log(x),
  (error: Error): void => console.log(error),
  () => console.log('completed')
);
// 結果
// completed

throwError

エラーを流します。

src/app.ts
/*
* throwError
* エラーを流します。
* throwError(error: any, scheduler?: SchedulerLike): Observable<never>
*/

of(1, 2, 3, 0).pipe(
  mergeMap((x: number): any => x === 2
    ? throwError('Thirteens are bad')
    : x
  )).subscribe(
  (x: number): void => console.log(x),
  (error: Error): void => console.log('error')
);

combineLatest

複数のObservableを受け取り他のObservableの直近のデータも合わせて流します。

src/app.ts/*
* combineLatest
* 複数のObservableを受け取り他のObservableの直近のデータも合わせて流します。
* combineLatest<T, R>(...observables: Array<any | ObservableInput<any> | Array<ObservableInput<any>> | (((...values: Array<any>) => R)) | SchedulerLike>): Observable<R>
*/

combineLatest([1, 5, 10].map(
  (n: number): ObservableInput<number[]> => of(n).pipe(
    delay(n * 1000),
    startWith(0),
  )
)).subscribe(<T>(array: T): void => console.log(array));

// 結果
// [0, 0, 0] immediately
// [1, 0, 0] after 1s
// [1, 5, 0] after 5s
// [1, 5, 10] after 10s