RxJS Promise Composition (передача данных)

Я совершенно новичок в Rx и нахожу трудным найти документацию по составлению обещаний, чтобы данные из первого обещания передавались во второе и так далее. Вот три очень простых обещания, расчеты по данным не важны, просто что-то асинхронное должно быть сделано с использованием данных из предыдущего обещания.

 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);

Моя любимая реализация-композит Рамды и pipeP: 

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

Похоже, что Rx может справиться с такой ситуацией довольно свободно. Однако самое близкое, что я нашел до сих пор, - это сравнение RxJS с async (библиотекой) здесь 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.Заметный.fromPromise (p1) не будет работать так, как он ожидает обещание само по себе не функция, а Rx.Заметный.defer (p1), похоже, не передает параметры, как в Примере.

Спасибо!

Аналогичный вопрос, но без передачи данных: связывание обещаний с RxJS

1 15
javascript functional-programming rxjs frp

1 ответ:

Я не читал все это, но если вы хотите достичь того же, что и pl().then(p2).then(p3).then(console.log);, с p будучи функцией, возвращающей обещания, вы могли бы сделать что-то вроде (пример здесь ) 

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);

Теперь, если вы хотите выполнить последовательный обратный вызов, обернутый через fromCallback или fromNodeCallback, Вы можете сделать что-то вроде: 

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)
Последний код непроверен, так что держите меня в курсе, если это сработает. Последний комментарий, это должно работать, если в каждом пункте у вас есть только одна ценность произведена (как обещание). Если у вас будет несколько файлов вместо одного, с flatMap Вы можете получить проблемы с заказом (если заказ имеет значение для вас), поэтому в этом случае вы можете использовать concatMap в качестве замены.
19