rx.angular.jsでインクリメンタルサーチ #JavaScript

先日、AngularJSの非同期処理をRxJSにつなげてみるという記事を書いたのですが、その数日後にReactive Extensionsの公式リポジトリにこんなのが作られてました。

rx.angular.js - Reactive Extensions Bindings for AngularJS

AngularJSとRxJSを連携させるためのライブラリのようです。
ソースコードを覗いてみましたが、現状では下記の2つの関数が用意されていました。

$toObservable
observeOnScope

どちらも$scopeの保持している変数の変更通知（すなわち$scope.$watch）をRxのObservableに変換するための機能です。
$scope.$toObservable('xxx')のように$scopeのメソッドとして呼び出すか、observeOnScope($scope, 'xxx')のように関数の引数に$scopeを渡すかが違うようですね。

さて、そのサンプルコードが面白かったのでご紹介。

app.js

$scope.search = '';
$scope.results = [];

var search = function(term) {
  var deferred = $http({
    url: "http://en.wikipedia.org/w/api.php?&callback=JSON_CALLBACK",
    method: "jsonp",
    params: {
      action: "opensearch",
      search: term,
      format: "json"
    }
  });

  return Rx.Observable
    .fromPromise(deferred)
    .select(function(response){
      return response.data[1];
    });
};

$scope
    .$toObservable('search')
    .throttle(300)
    .map(function(data){
        return data.newValue;
    })
    .distinctUntilChanged()
    .select(search)
    .switchLatest()
    .subscribe(function(val){
        $scope.results = val;
    });

これは、サーバーに問い合わせるタイプのインクリメンタルサーチのサンプルです。
$scope.searchがテキストボックスにバインドされていて、入力した文字列をWikipediaで調べて、その結果を$scope.resultsに出力するようになっています。

前半の処理は、$httpでWikipediaに問い合わせをして、その結果のpromiseをRxJSで扱えるように変換しています。
面白いのは後半の処理なので、1行ずつ解説します。

$scope.$toObservable('search')
- $scope.searchの変更通知をRxJSで扱えるように変換します。
throttle(300)
- 指定した時間（300msec）、新たな変更通知が発生しなかったら最後に発生した値を後続に流します。
- テキストボックスに連続で文字を入力するときに、1文字入力するたびにサーバーに問い合わせるのはうれしくないですからね。
map(・・・)
- dataはnewValueとoldValueを持っているので、newValueだけ後続に流します。
distinctUntilChanged()
- 前回と値が同じだったら後続に流しません。
select(search)
- $httpを使ってWikipediaに問い合わせを行います。
switchLatest()
- 同じタイミングでサーバーへの問い合わせが複数発生した場合、最後の問い合わせの結果だけを後続に流します。
- throttle()は時間的に最後に発生した値を後続に流すものでしたが、こちらは最後の問い合わせの結果を後続に流します。最後に問い合わせたものが、時間的には先に返ってくることもありますからね。
subscribe(・・・)
- $scope.resultsに結果を入れます。

リアクティブプログラミングは最初はなかなか取っ付きづらいですが、慣れてくると複雑な処理がこんなにすっきり書けます。便利ですね！