RxJSプロミス組成（データ受け渡し）

Question

私はRxの新作で、最初の約束事からのデータが2番目に渡されるような約束事に関する文書を見つけるのが難しいと感じています。ここでは3つの基本的な約束事がありますが、データの計算は重要ではありません。前の約束のデータを使用して非同期処理を行わなければならないということだけです。私が話している構図を達成するために

const p1 =() => Promise.resolve(1); 
const p2 = x => { const val = x + 1; return Promise.resolve(val); }; 
const p3 = x => { 
     const isEven = x => x % 2 === 0; 
     return Promise.resolve(isEven(x)); 
}; 

伝統的な方法：

pl().then(p2).then(p3).then(console.log); 

私のお気に入りの実装はRAMDAのcomposePとpipePです：

R.pipeP(p1, p2, p3, console.log)() 

Rxができるかもしれません可能性が高いと思われますこの種の状況をかなり流暢に扱うことができます。しかし、私がこれまでに発見した最も近いRxJSから非同期（ライブラリ）の比較にあるここhttps://github.com/Reactive-Extensions/RxJS/blob/master/doc/mapping/async/comparing.md：

var Rx = require('rx'), 
    fs = require('fs'), 
    path = require('path'); 
var file = path.join(__dirname, 'file.txt'), 
    dest = path.join(__dirname, 'file1.txt'), 
    exists = Rx.Observable.fromCallback(fs.exists), 
    rename = Rx.Observable.fromNodeCallback(fs.rename), 
    stat = Rx.Observable.fromNodeCallback(fs.stat); 
exists(file) 
    .concatMap(function (flag) { 
    return flag ? 
     rename(file, dest) : 
     Rx.Observable.throw(new Error('File does not exist.')); 
    }) 
    .concatMap(function() { 
     return stat(dest); 
    }) 
    .forEach(
     function (fsStat) { 
      console.log(JSON.stringify(fsStat)); 
     }, 
     function (err) { 
      console.log(err); 
     } 
    ); 

concatMapは有望なようだが、上記のコードはかなり恐ろしい見えます。 Rx.Observable.fromPromise（p1）は関数ではなく約束そのものを期待しており、Rx.Observable.defer（p1）は次のようなパラメータを渡すようには見えないので、私の例では問題がありました。例。

ありがとう！

同様の質問が、データの受け渡しなし： Chaining promises with RxJS

Answer 1

私はそれのすべてを読んでいないが、あなたはpl().then(p2).then(p3).then(console.log);と同じを達成する必要がある場合は、pは約束を返す関数であることを、あなたは（例のようなものを行うことができますhere）

Rx.Observable.fromPromise(p1()) 
      .flatMap(function(p1_result){return p2(p1_result);}) 
      .flatMap(function(p2_result){return p3(p2_result);}) 

以上の対称性：

var chainedPromises$ = 
    Rx.Observable.just() 
      .flatMap(p1) 
      .flatMap(p2) 
      .flatMap(p3); 

function rename (flag){ 
    return flag 
      ? rename(file,dest).flatMap(return Rx.Observable.just(dest)) 
      : Rx.Observable.throw(new Error('File does not exist.')); 
} 

Rx.Observable.just(file) 
      .flatMap(exists) 
      .flatMap(rename) 
      .flatMap(stat) 

後者のコードがテストされていないので、それが動作するかどうか、私は最新の状態に保つ：あなたは順次fromCallbackまたはfromNodeCallbackによって包まれたコールバックを実行したい場合は今、あなたのような何かを行うことができます。 最後のコメントです。これは、それぞれのポイントで（プロミスのように）1つの価値しか生まれない場合に有効です。複数のファイルを1つではなくflatMapとすると、注文の問題が発生する可能性があります（注文に問題がある場合）。その場合はconcatMapを代わりに使用できます。