RxJS Promise Composition (truyền dữ liệu)

Question

Tôi là người mới sử dụng Rx và thấy khó tìm tài liệu về các lời hứa soạn thảo để dữ liệu từ lời hứa đầu tiên được chuyển sang lần thứ hai và cứ tiếp tục như vậy. Dưới đây là ba lời hứa rất cơ bản, các tính toán trên dữ liệu không quan trọng, chỉ cần một cái gì đó không đồng bộ phải được thực hiện bằng cách sử dụng dữ liệu từ lời hứa trước đó.

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

Cách truyền thống để đạt được các thành phần Tôi đang nói về:

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

thực ưa thích của tôi là Ramda của composeP và pipeP:

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

Dường như Rx có thể có thể để xử lý tình huống này khá trôi chảy. Tuy nhiên, gần nhất tôi đã tìm thấy cho đến nay là từ RxJS để async (thư viện) so sánh ở đây 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 dường như hứa hẹn, nhưng mã trên có vẻ khá khủng khiếp. Tôi cũng gặp rắc rối với ví dụ của tôi vì Rx.Observable.fromPromise (p1) sẽ không hoạt động vì nó mong đợi một lời hứa, không phải là một hàm, và Rx.Observable.defer (p1) dường như không truyền các tham số như thí dụ.

Cảm ơn!

câu hỏi tương tự nhưng không có dữ liệu qua: Chaining promises with RxJS

Answer 1

Tôi đã không đọc tất cả của nó, nhưng nếu bạn muốn đạt được giống như pl().then(p2).then(p3).then(console.log);, với p là chức năng hứa hẹn trở lại, bạn có thể làm điều gì đó tương tự (ví dụ here)

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

Hoặc đối xứng hơn:

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

Bây giờ nếu bạn muốn thực hiện tuần tự gọi lại quấn qua fromCallback hoặc fromNodeCallback, bạn có thể làm một cái gì đó như:

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) 

Mã thứ hai là chưa được kiểm tra, vì vậy giữ cho tôi cập nhật nếu mà làm việc. Nhận xét cuối cùng, điều này sẽ hoạt động nếu tại mỗi điểm bạn chỉ có một giá trị được tạo ra (như lời hứa). Nếu bạn có nhiều tệp thay vì một, với flatMap bạn có thể nhận được các vấn đề đặt hàng (nếu đơn hàng quan trọng với bạn), vì vậy trong trường hợp đó, bạn có thể sử dụng concatMap làm phương án thay thế.