将一系列有序操作组合成一个 Observable答案

【问题标题】：Combine a list of ordered operations into one Observable将一系列有序操作组合成一个 Observable
【发布时间】：2016-10-24 21:58:17
【问题描述】：

var state = [];
var operation1 = function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state.push(1, 2);
        setTimeout(resolve, 300, state);
    }));
};
var operation2 = function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.map(x => x * 2);
        setTimeout(resolve, 200, state);
    }));
};
var operation3 = function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.reduce( (prev, next) => prev + next );
        setTimeout(resolve, 100, state);
    }));
};
var operations = [operation1, operation2, operation3];

鉴于上面的代码，我试图将operations 组合成一个Observable，它会发出每个操作的状态。所以Observable 需要做以下任一操作：

发射3次：[1, 2], [2, 4], 6
发射1次：[[1, 2], [2, 4], 6]

【问题讨论】：

你的意思是说你只想在链中的前一个promise完成后才做产生promise的工作吗？
@GregL 是的，这是正确的。我已经更新了示例以显示这一点。

标签： javascript promise rxjs observable rxjs5

【解决方案1】：

你可以试试 (jsbin)

var state = [];
var operation1 = Rx.Observable.defer(function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state.push(1, 2);
        setTimeout(resolve, 300, state);
    }));
});
var operation2 = Rx.Observable.defer(function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.map(x => x * 2);
        setTimeout(resolve, 200, state);
    }));
});
var operation3 = Rx.Observable.defer(function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.reduce( (prev, next) => prev + next );
        setTimeout(resolve, 100, state);
    }));
});
var operations = Rx.Observable.from([operation1, operation2, operation3]).merge(1);
operations.subscribe(function(x){console.log(x)})

请检查这是否有效，稍后我将详细说明这是如何工作的。

【讨论】：

谢谢！这个解决方案很简单，效果很好！关键是在这里使用Rx.Observable.defer。

【解决方案2】：

您可以使用Rx.Observable.concat，但我认为如果没有 rx.java 并使用 Promise.all，这个简单的案例会更容易

let slow = new Promise((resolve) => {
  setTimeout(resolve, 200, 'slow');
});
let instant = new Promise((resolve) => {
  setTimeout(resolve, 0, 'instant');
});
let quick = new Promise((resolve) => {
  setTimeout(resolve, 50, 'quick');
});

var operation1 = function() {
  return Rx.Observable.fromPromise(slow);
}
var operation2 = function() {
  return Rx.Observable.fromPromise(instant);
}
var operation3 = function() {
  return Rx.Observable.fromPromise(quick);
}
var operations = [operation1(), operation2(), operation3()];
var source = Rx.Observable.concat(operations);

var subscription = source.subscribe(
  function(x) {
    console.log('Next: ' + x);
  },
  function(err) {
    console.log('Error: ' + err);
  },
  function(a) {
    console.log('Completed', a);
  });

// Or with promises

operations = [slow, instant, quick]

Promise.all(operations)
.then(console.log.bind(console, "Promise all"))

&lt;script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.all.min.js"&gt;&lt;/script&gt;

【讨论】：

问题是我不能按顺序调用这些操作。 Promise.resolve() 仅用于演示目的。我真正的承诺将延迟解决。而且我想确保在调用下一个承诺之前解决第一个承诺。
Promise.all 将按照承诺在列表中的顺序解决所有问题。有没有延迟也没关系
延迟以及接下来解决承诺的顺序对我来说很重要。 Promise.all 不会确保数组中的第一个承诺在下一个开始解析之前被解析。我更新了我的示例代码，以更好地说明为什么顺序和延迟很重要。
如果你真的需要，首先你可以使用Rx.Observable.concat。其次，您可以查看此处以更好地理解承诺。 stackoverflow.com/questions/28066429/…
@denixtry 传递给 Promise.all 的承诺并行解决 - 不是按照传递的顺序。