javascript - ReactiveX 对 observable 进行多次过滤并合并

Question

我在创建以下可观察值时遇到问题。
我希望它接收预定义的值数组
我想按不同的事物进行过滤，并能够将它们作为单独的可观察值进行处理。
然后，当需要合并这些过滤后的可观察值时，我想保留原始观察值的顺序

//Not sure the share is necessary, just thought it would tie it all together
const input$ = Observable.from([0,1,0,1]).share();
const ones$ = input$.filter(n => n == 1);
const zeroes$ = input$.filter(n => n == 0);

const zeroesChanged$ = zeroes$.mapTo(2);
const onesChanged$ = ones$.mapTo(3);
const allValues$ = Observable.merge(onesChanged$,zeroesChanged$);

 allValues$.subscribe(n => console.log(n));
//Outputs 3,3,2,2
//Expected output 3,2,3,2

编辑:很抱歉我的问题不够具体。我正在使用一个名为cycleJS 的库，它将副作用分离为驱动程序。所以我在我的周期中所做的是这样的

export function socketCycle({ SOCKETIO }) {
  const serverConnect$ = SOCKETIO.get('connect').map(serverDidConnect);
  const serverDisconnect$ = SOCKETIO.get('disconnect').map(serverDidDisconnect);
  const serverFailedToConnect$ = SOCKETIO.get('connect_failed').map(serverFailedToConnect);
  return { ACTION: Observable.merge(serverConnect$, serverDisconnect$, serverFailedToConnect$) };
}

现在，当我想为其编写测试时，我的问题出现了。我尝试了以下方法，结果出现了错误(使用笑话)

const inputConnect$ = Observable.from(['connect', 'disconnect', 'connect', 'disconnect']).share();
const expectedOutput$ = Observable.from([
  serverDidConnect(),
  serverDidDisconnect(),
  serverDidConnect(),
  serverDidDisconnect(),
]);
const socketIOMock = {
  get: (evt) => {
    if (evt === 'connect') {
      return inputConnect$.filter(s => s === 'connect');
    } else if (evt === 'disconnect') {
      return inputConnect$.filter(s => s === 'disconnect');
    }
    return Observable.empty();
  },
};
const { ACTION } = socketCycle({ SOCKETIO: socketIOMock });
Observable.zip(ACTION, expectedOutput$).subscribe(
  ([output, expectedOutput]) => { expect(output).toEqual(expectedOutput); },
  (error) => { expect(true).toBe(false) },
  () => { done(); },
);

也许我可以通过另一种方式来测试它？

Answer 1

当流被分区时，不同子流中元素之间的时序保证实际上被破坏了。特别是，即使 connect 事件始终出现在事件源处的 disconnect 事件之前，connect Observable 的事件也不会总是出现在之前disconnect Observable 中相应的事件项。在正常时间范围内，这种竞争情况可能相当罕见，但仍然很危险，并且此测试显示了最坏的情况。

好消息是，所示的函数只是事件和处理程序结果之间的映射器。如果您可以在事件类型上继续使用此模型，那么您甚至可以在普通数据结构中对映射进行编码，这有利于表达:

const event_handlers = new Map({
 'connect': serverDidConnect,
 'disconnect': serverDidDisconnect,
 'connect_failed': serverFailedToConnect
});
const ACTION = input$.map(event_handlers.get.bind(event_handlers));

警告:如果您要减少子流(或以其他方式考虑以前的值，例如 debounceTime)，则重构并不那么简单，并且还取决于关于“维持秩序”的新定义。大多数时候，使用reduce+一个更复杂的累加器来重现仍然是可行的。