Angular 1.x + RxJS でいい感じに subscribe する


Angular 2 には RxJS 5 が組み込まれていますが、Angular 1.x 用にも [rx.angular.js](https://github.com/Reactive-Extensions/rx.angular.js) というライブラリがあり、scope の変更や click などのイベントハンドラを簡単に `Observable` にすることができ大変便利です。

しかし、Angular 1.x で RxJS を使う際に、いくつか気をつけなければいけない点があります。

- `$destroy` 時の `dispose()`
- digest loop の開始

これらを各所で考えながら使うのは面倒なので、何も考えずに `subscribe()` の代わりに使えるメソッドを考えてみました。

## `$destroy` 時の `dispose()`

controller（や directive の controller）で `subscribe()` をする場合、通常 scope の `$destroy` イベントを受けて `Subscription#dispose()` で `subscribe()` をキャンセルする必要があります。そうしないと、controller がなくなった後も stream が生き続けるゾンビ状態になってしまうためです。

`subscribe()` が返す `Subscription` を引き回す必要があり、複数の `subscribe()` がある時などなかなか面倒です。

```js
const fooSub = foo$.subscribe(x => doSomething(x));
const barSub = bar$.subscribe(x => doOtherThing(x));

$scope.$on('$destroy', () => {
  fooSub.dispose();
  barSub.dispose();
});
```

そこで [Ben Lesh さんの RxJS: Don’t Unsubscribe](https://medium.com/@benlesh/rxjs-dont-unsubscribe-6753ed4fda87#.tkmznu7ho) を読んで気づいたのですが [`takeUntil()`](https://github.com/Reactive-Extensions/RxJS/blob/master/doc/api/core/operators/takeuntil.md) と [`$eventToObservable()`](https://github.com/Reactive-Extensions/rx.angular.js/tree/master/docs#eventtoobservableeventname) を使うとオシャレに stream を止めることができます。`$destroy` イベントを `Observable` にして `takeUntil` に入れることで、`$destroy` イベントが来た時に stream を `completed` にして止めることができます。`completed` になった stream はそれ以上 `dispose()` する必要がありません。

```js
const destroy$ = $scope.$eventToObservable('$destroy');

foo$.takeUntil(destroy$).subscribe(x => doSomething(x));
bar$.takeUntil(destroy$).subscribe(x => doOtherThing(x));
```

このパターンは Angular 1.x に限らず、コンポーネントの破棄イベントを `Observable` にできれば他のフレームワークでも使えそうです。

## digest loop の開始

click や HTTP レスポンスなどのイベントならば、その後 digest loop が走りますが、Web Socket からのイベントや、タイマー的なものを使っている場合そうならないケースが出てきます。`subscribe()` のコールバック内で毎回 `$scope.$apply()` を呼ぶのも面倒です。[ScopeScheduler](https://github.com/Reactive-Extensions/rx.angular.js/tree/master/docs#rxscopescheduler) というものもありますが、たまに忘れてしまいそうです。

[rx.angular.js](https://github.com/Reactive-Extensions/rx.angular.js) は [`safeApply()`](https://github.com/Reactive-Extensions/rx.angular.js/tree/master/docs#safeapplyscope-fn) という便利なオペレーターを用意してくれています。ただこれは `subscribe()` はしてくれない `tap()` 的なものなので、`subscribe()` と組み合わせて使う必要があります。

```js
something$.safeApply($scope, x => doSomething(x)).subscribe();
```

[`safeApply()` のドキュメント](https://github.com/Reactive-Extensions/rx.angular.js/tree/master/docs#safeapplyscope-fn)には書いていないのですが、第 3 引数、第 4 引数にそれぞれ `onError`, `onCompleted` のコールバックを渡すことができます。これで `subscribe()` の代わりになりそうですね。

しかし、実は `safeApply()` は `tap()` を元にした実装になっているため、`subscribe()` の代わりになりません。`onError` を設定しても `subscribe()` の第 2 引数を指定していなければ error が流れてきた時に例外が投げられてしまうのです。

この問題があるので `safeApply()` の `tap()` ではなく `subscribe()` バージョンが欲しくなります。

Angular 2 なら [Zone.js](https://github.com/angular/zone.js/) のおかげでこういうことをする必要はないはずです。また Async Pipe を使えば自分で `subscribe()` を書く必要もありません。

## まとめると

以下のような実装だと何も考えずに `subscribe()` の代わりとして使えるのではないかと思います。

```js
something$.safeSubscribe($scope, x => doSomething, e => showError(e));
```

```js
angular.module('rx.safeSubscribe', ['rx']).run((rx) => {
  rx.Observable.prototype.safeSubscribe = function (scope, onNext, onError, onCompleted) {
    return this
      .takeUntil($scope.$eventToObservable('$destroy'))
      .subscribe(
        onNext && safeify(onNext),
        onError && safeify(onError),
        onCompleted && safeify(onCompleted)
      );
  };

  function safeify(scope, func) {
    return x => {
      if (scope.$$phase || scope.$root.$$phase) {
        func(x);
      } else {
        scope.$apply(() => func(x));
      }
    }
  }
});

```

もっと良い方法があれば教えてください！