新卒のMです。

ReactiveExtensions(Rx)
RxはC#でLINQが作られたことから始まり、
その利便性からLINQを各種言語へRxとして移植までされているほどです。

※LINQについては今回は詳しく説明しません。簡単に言うとデータをクエリのようなもので扱うものです。

Rxは簡単に説明すると、イベントを取り扱うものです。
この時条件式は自由に設定できますし、途中から変更したりもできます。

今回はJavaScript(JS)へと移植された、RxJSを使ってみました。

まずは簡単に使ってみます。

Rx.Observable.interval(1000)
    .subscribe(function(cnt){console.log("経過時間:"+cnt+"秒");});

これで経過秒数をカウントし、出力するプログラムの完成です。
.intervalを付けることで、指定したms毎に条件を満たすので、subscribeされます。
ここに、終了フラグを付けてみます。

//  ストップウォッチ
Rx.Observable.interval(1000)
    .filter(function(){return !endFlag;})
    .subscribe(function(cnt){console.log("経過時間:"+cnt+"秒");});
//  タイマー
function(time){
    Rx.Observable.interval(1000)
        .take(time+1)
        .subscribe(function(cnt){console.log("残り時間:"+(time-cnt)+"秒");});
}

これでフラグがたてば止まるストップウォッチと、
関数に投げた時間で止まるタイマーができました。
filterを付けることで、intervalが満たされた時にfilterの判定が入りtrueが返された時にだけsubscribeされます。

createを使うと複雑なものも作れます。

var source = Rx.Observable.create(function(observer){
    //  1秒ごとに終了条件を確認
    var interval = setInterval(function(){
        if( endFlag ){
            //  onCompleted()を呼ぶことでストリームを終了
            //  終了後disposeされる
            observer.onCompleted();
        }
    }, 1000);
    //  disposeされたときに呼ばれる
    return function(){
        //  後始末 
        clearInterval(interval);
        console.log('disposed');
    }
});

//  各関数ごとの処理を追加
var subscription = source.subscribe(
        //  onNext(x)時の処理
        function(x){
            console.log('onNext:' + x);
        },
        //  onError(e)時の処理
        function(e){
            console.log('onError');
        },
        //  onCompleted()時の処理
        function(){
            console.log('onCompleted');
        }
);

これまでの書き方だとストリームでのイベント処理は行えますが、
自分自身でストリームを消すことができません。
ですが、上記の書き方だとストリームの終了まで
そのストリーム自身で行うことができます。
もちろんこのままでは使い道はありませんが、
その時に応じて幅広く使うことができます。

導入はとても簡単でした。
https://github.com/Reactive-Extensions/RxJS
ここからrx.all.jsをダウンロードして、
スクリプトを読み込ませるだけです。

このライブラリの利点は時間を使った処理や、
イベント処理を独立して書くことができる点です。

UnityなどのC#などでも使えるので興味が出たら触ってみましょう。

以上、簡単にRxJSを触ってみました。