合并可观察序列,其中一些是异步并保留顺序

Question

作为一个例子,我有两个可观察的序列,"data1"和"data2",我希望将它们合并为一个可观察序列,同时保留初始顺序.

当两个可观察序列没有异步工作时 - 这里以小延迟建模 - 这很容易实现Rx.Observable.merge.但是,使用这种方法,任何异步工作都会破坏订单.

是否可以合并可观察序列,使用内置运算符传递已知和等待值尚未知的值？如果没有,我应该建立什么类型的运营商？

'use strict';
var Rx = require('rx');
var EventEmitter = require('events').EventEmitter;

var eventEmitter = new EventEmitter();

var end = Rx.Observable.fromEvent(eventEmitter, 'end');

var data1 = Rx.Observable.fromEvent(eventEmitter, 'data1')
  .takeUntil(end);
var data2 = Rx.Observable.fromEvent(eventEmitter, 'data2')
  .takeUntil(end)
  .delay(1000);

Rx.Observable
  .merge(data1, data2)
  .reduce(function(acc, str) {
    return acc + str + ';';
  }, '')
  .subscribe(function(data) {
    console.log(data);
  });

eventEmitter.emit('data1', '1');
eventEmitter.emit('data2', '2');
eventEmitter.emit('data1', '3');
eventEmitter.emit('data1', '4');
eventEmitter.emit('data2', '5');
eventEmitter.emit('end');

// expected: "1;2;3;4;5;"
// actual: "1;3;4;2;5;"

谢谢.

Answer 1

一个选项:在完成工作之前合并流,以便捕获订单,然后完成工作(使用concat维护订单)

// synchronous work...but return it as an Observable so that
// concatMap() can use it correctly
var data1Work = function (data) { return Rx.Observable.of(data); };
// asynchronous work...returns an observable with the result
var data2Work = function (data) {
    // cold observable (wont start until concatMap hits it)
    var work = Rx.Observable.of(data).delay(1000);
    // return work;

    // return hot observable (starts immediately)
    hotwork = work.replay();
    hotwork.connect(); // start it now
    return hotwork;


    // alternatively you can start your async operation and return
    // a Promise that will resolve when it is complete if that
    // pattern feels better to you since Promises are always hot
    // and Rx knows how to consume promises
    // return someFuncThatReturnsPromise(data);
};

var data1 = Rx.Observable.fromEvent(eventEmitter, "data1")
    .map(function (d) { return { type: "data1", data: d }; });
var data2 = Rx.Observable.fromEvent(eventEmitter, "data2")
    .map(function (d) { return { type: "data1", data: d }; });

var results = Rx.Observable
    .merge(data1, data2)
    .concatMap(function (d) {
        var workFunc = d.type === "data1" ? data1Work : data2Work;
        return workFunc(d.data);
    });