ちきさんです。

Scalaのfor yieldみたいなことをRxJSで表現できないかと思ってやってみました。

ひとまとめの複数の非同期処理のうち、一つでもエラーになったら他の全ての非同期処理を破棄するRxJSのコードです。

デモ → http://jsbin.com/xerufem/10/edit?js,console

エラーが一つもなければ出力に HELLO と表示されますが、そうでなければ (empty) と表示されます。ランダムに結果が変わるので何回かRUNボタンを押して動作を確認してみてください。

以下、JS Binのコードに少しコメントを足したものです。

ScalaのSomeとNoneに相当するものがObservable.ofとObservable.emptyになります。

const Observable = Rx.Observable

// 9割の確率で成功する。
function isSuccess() {
  return Math.random() > 0.1
}

// 非同期処理。たまに失敗する。成功すると文字を大文字にして返す。
// 成功のときはObservable.of(value)を返すが、失敗のときはObservable.emptyを返す。
function httpRequest(value) {
  const promise = new Promise((resolve, reject) => {
    if (isSuccess()) {
      resolve(value.toUpperCase())
    } else {
      reject()
    }
  })
  // Observable.fromはPromiseをObservableに変換する。
  return Observable.from(promise) // resolveのときはObservable.ofを返す。
    .catch(_ => Observable.empty()) // rejectのときはObservable.emptyを返す。
}

// 複数の非同期処理をひとまとめにする。
function requests() {
  return [
    httpRequest('h'),
    httpRequest('e'),
    httpRequest('l'),
    httpRequest('l'),
    httpRequest('o'),
  ]
}

// ひとまとめの非同期処理を10回走らせる。理論上半分ぐらいは失敗する。
for (let i = 0; i < 10; i++) {
  // Observable.forkJoinはPromise.allに似た動きをするが、一つでもemptyを返すものがあるとemptyを返す。 
  Observable.forkJoin(requests())
    .map(array => array.join(''))
    .defaultIfEmpty('(empty)') // Observable.emptyのときにセットする値。
    .zip(Observable.of(i)) // ストリームに外側の変数を巻き込む。
    .map(([value, index]) => index + ': ' + value)
    .subscribe(console.log)
}