Angularの中でオブザーバとRxJSが重要な機能です。使い方はある程度分かるのが実情で実体は何という人が多い。Unix環境などを知っているはデータストリーミングの考え方があるのはご存知でしょうが、プログラミングの人には理解不能になります。
特にバージョン5から6になり、RxJSの書き方などが一部変わり、またソースチェックが厳しくなり、以前と違い、エラーを起こし、大混乱をしています。
今後バージョン7が正式になると、オブザーバとRxJSはさらに理解度を増さないとロジックは書けなくなります。
今回はRxJSとオブザーバに焦点を当てます。
Contents
オブザーバとは
Observable
Observableは、アプリケーションの中でパブリッシャーとサブスクライバー間でメッセージを渡すためのサポートを提供します。Observableは、イベント処理、非同期プログラミング、および複数の値の処理のための他のテクニックよりも大きな利点を提供します。Observableは、文脈に応じて、任意の型—リテラル、メッセージ、またはイベントの複数の値を提供できます。受け取るためのAPIは値が同期的・非同期的に提供される場合も同じです。基本的なセットアップとティアダウンはObservableにとって処理されるので、あなたのアプリケーションコードは値を消費するためにサブスクライブを行うことと、それが済んだらアン購読することだけを心配する必要があります。ストリームがキー入力、HTTPレスポンス、インターバルタイマーのどれでも、値をリスニングしたり、リスニングを止めるためのインターフェースは同じです。
これらの利点のために、ObservableはAngular内で広く使用されており、アプリの開発にも推奨されています。
基本的な使用法と用語
パブリッシャーとして、サブスクライバー 関数を定義する Observable インスタンスを作成することができます。これは、コンシューマーが subscribe（）メソッドを呼び出したときに実行される関数です。サブスクライバー関数は、どのように値を取得または生成し、メッセージを発行するかを定義します。作成したObservableを実行して値を受信するには、オブザーバー を渡す subscribe() メソッドを呼ぶ必要があります。これは、受け取った通知のハンドラーを定義するJavaScriptオブジェクトです。subscribe() は 通知を受信を止めるための unsubscribe() メソッドがある Subscription オブジェクトを返します。
ドキュメント引用しましたが、用語辞典が必要になります。
要約すれば、オブザーバは、データ(配列やオブジェクトなど各種のデータ)をメッセージとして取り扱い、メッセージを作る送り手側、メッセージを受けて側で取り扱い方を簡単にするのが目的です。
例えばHTTP要求をし、HTTP応答を受け取る際にオブザーバが利用されています。
エラー処理の仕方など、簡単に書けること、同期、非同期に関係なく、同じ処理になるメリットがあります。
最終的にはメンテンナス性が良く、ロジックやテストをする際にも容易になります。
※オブザーバの基本技術は、Pub/Subメッセージングモデルを利用しています。
メッセージを作成して送信する送信側のクライアント（プロデューサー）をパブリッシャーといいます。また，メッセージを受信する側のクライアント（コンシューマー）をサブスクライバーといいます。
オブザーバの定義
Observableの通知を受け取るハンドラーは、Observer インターフェースを実装します。これは、Observableが送信できる3種類の通知を処理するためのコールバックメソッドを定義するオブジェクトです。
next |
必須です。個々の値が提供されたときのハンドラーです。実行が開始されてから0回以上呼び出されます。 ここに実装すべき処理が入ります。 |
error |
オプションです。エラー通知のハンドラーです。エラーはObservableインスタンスの実行を停止します。 |
complete |
オプションです。実行完了通知のハンドラーです。遅延した値は、実行完了後もnextハンドラーに引き続き渡されます。 |
※見本などの説明は、RxJSと一緒に説明します。
サブスクライブ
Observable インスタンスは誰かが購読すると値をパブリッシュしはじめます。購読するためにはインスタンスの subscribe() メソッドを呼び出し、オブザーバーオブジェクトを渡して通知を受け取ります。
※見本などの説明は、RxJSと一緒に説明します。
RxJSとは
リアクティブプログラミングは、データストリームと変更の伝播 に関する非同期プログラミングのパラダイムです。RxJS (Reactive Extensions for JavaScript) は、非同期またはコールバックベースのコード (RxJS Docs) の作成を容易にする observables を使用したリアクティブプログラミング用のライブラリです。
RxJS は Observable 型の実装を提供します。Observable 型は、型が言語の一部となるまで、そしてブラウザがそれをサポートするまで必要です。ライブラリはまたobservablesを作成して作業するためのユーティリティ関数を提供します。これらのユーティリティ関数は、次の用途に使用できます。
- 非同期処理の既存のコードを observables に変換する
- ストリーム内の値を反復処理する
- 異なる型への値のマッピング
- ストリームのフィルタリング
- 複数のストリームの作成
- Observable 作成関数
RxJS には新しい observables を作成するために使用できるいくつかの関数が用意されています。これらの関数はイベント、タイマー、promise などから observables を作成するプロセスを簡素化できます。
引用文は難しい。要約的に言えば、データ処理はRxJSを用いてオブザーバを使いますということです。イベント、タイマー、promiseからオブザーバに変換する関数も用意されていますということです。
特にバージョン5までと6以降はオブザーバとRxJSの取り扱いが変わり、戸惑う人が増えています。ドキュメントで必要なことを読みなさいと言って、ドキュメントを読みこなすことは難しいのが現状です。
次に必要なことをピックアップします。
promise から observable を作成する
es2015やes2017仕様では promise を用いる人が多い。
これをオブザーバに置き換えないと、思うわぬエラーが出るので、下記のようにfromPromise を用いてオブザーバに変換します。
import { fromPromise } from 'rxjs';
// promise から observable を作成する
// HTTP通信部分はPromiseオブジェクトが利用されています。
const data = fromPromise(fetch('/api/endpoint'));
// 非同期の結果を購読します。
data.subscribe({
next(response) { // 個々の値が提供されたときのハンドラーの処理 },
error(err) { // エラー通知のハンドラーの処理},
complete() { // 実行完了の通知ハンドラーの処理 }
});
カウンターから observable を作成する
import { interval } from 'rxjs';
// interval はobservable を作成します
const secondsCounter = interval(1000);
// 値を購読します。
secondsCounter.subscribe(n => {
console.log(`It's been ${n} seconds since subscribing!`));
イベントから observable を作成する
import { fromEvent } from 'rxjs';
const el = document.getElementById('my-element');
// エレメントのオブジェクトは Observable として作成されます。
const mouseMoves = fromEvent(el, 'mousemove');
// 上記のイベントを購読します。
const subscription = mouseMoves.subscribe((evt: MouseEvent) =<span class="pun">></span> <span class="pun">{</span>
// マウスのXY軸をコンソールログに出力します
console.log(`Coords: ${evt.clientX} X ${evt.clientY}`);
// When the mouse is over the upper-left of the screen,
// 非購読をし、処理が終了します。
if (evt.clientX < 40 && evt.clientY < 40) {
subscription.unsubscribe();
}
});
ユーザ登録の場合のソース見本
ユーザ登録のコンポーネントです。
import { Component, OnInit } from '@angular/core';
import { Router } from '@angular/router';
import { FormBuilder, FormGroup, Validators } from '@angular/forms';
import { first } from 'rxjs/operators';
import { AlertService, UserService } from '../_services';
@Component({ templateUrl: 'register.component.html' })
export class RegisterComponent implements OnInit {
registerForm: FormGroup;
loading = false;
submitted = false;
constructor(
private formBuilder: FormBuilder,
private router: Router,
private userService: UserService,
private alertService: AlertService) { }
ngOnInit() {
this.registerForm = this.formBuilder.group({
firstName: ['', Validators.required],
lastName: ['', Validators.required],
username: ['', Validators.required],
password: ['', [Validators.required, Validators.minLength(6)]]
});
}
// convenience getter for easy access to form fields
get f() { return this.registerForm.controls; }
onSubmit() {
this.submitted = true;
// stop here if form is invalid
if (this.registerForm.invalid) {
return;
}
this.loading = true;
this.userService.register(this.registerForm.value)
.pipe(first())
.subscribe(
data => {
this.alertService.success('登録が完了しました。', true);
this.router.navigate(['/login']);
},
error => {
this.alertService.error(error);
this.loading = false;
});
}
}
前提として、登録の際はOKならばステータスが200で戻りますし、エラーがあればエラー文言がサーバから戻ります。
43行目：購読を開始します。
44行目：常に配列として戻りますので、最初の1件のみを取得します。
45行目：登録が成功したのでメッセージを表示用です。
46行目：登録が完了したのでログイン画面に遷移させます。
47行目：サーバで項目などのチェックをした際のエラー処理の内容を表示させます。
例えば、ユーザが同一である場合などです。
48行目：HTTP応答に時間がかかる場合、ロード中を表示させるフラグですので説明は割愛します。
※この見本ではステータスが200以外は、インターセプタでエラー処理をしています。
補足事項
Angular6では以前のバージョンからUpdateが可能となりました。
ただし、複雑な処理のRxJSとオブザーバだとUpdate後、エラーが多発します。ソースの修正を自動的に行うので意外と思われるでしょうが、関係のないものまで削除します。
その後、修正作業は工数がかかります。まさかの削除でした。
簡単なものは良いですが、やはり実環境のアプリケーションでは、バージョン相違にある部分はチェックして修正したほうが良いです。
Updateを経験しましたが、意外と落とし穴が多く、その後の修正工数が膨大でした、Angular6で作り直すほうが工数が少ないと思われます。これはあくまで個人的見解です。
まとめ
最初に書きましたように、現在、バージョン6とバージョン5までの書き方が違い混乱する人が多いです。両方を取り扱っていると、思わぬ不具合が発生しますのでお気をつけてください。
Angular6で何をしているのか理解して、ロジック作成に取り掛かってください。