ReactiveX 对 observable 进行多次过滤并合并答案

【问题标题】：ReactiveX filtering on observable multiple times and mergingReactiveX 对 observable 进行多次过滤并合并
【发布时间】：2018-05-26 19:13:27
【问题描述】：

我在创建以下 observable 时遇到问题。
我希望它接收预定义的值数组
我想过滤不同的东西，并能够将它们作为单独的可观察对象使用。
然后当需要合并这些过滤后的 observables 时，我想保留原始的顺序

//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(); },
);

也许我可以通过另一种方式对其进行测试？

【问题讨论】：

input$.map(n => n + 2) 并且您应该更仔细地查看您所期望的输出。
如果您在每个 Observable 上的“工作”一次只考虑一个值，那么正如 @cartant 所建议的，这不应被视为“过滤、处理、然后合并”，而只是直接映射。在最一般的情况下，您的映射函数将使用 if-else 将值指向不同的行为。如果他们自己返回 Observables，您可以concatMap 保留您的订单。
实际上我已经将地图和合并部分分离到另一个函数（获取源并返回接收器），并且当输入异步时它工作正常，但是当我尝试测试它时，这里输入是事先指定的，它不保留顺序...
这个其他函数在做什么？如果它不是映射函数，那么您必须更具体地说明“保留顺序”的含义。
你不能说两个独立的流之间元素的相对时间，即使它们来自同一个流。你在这里看到了后果。不管怎样，您的测试总是会中断，因为它依赖于因拆分原始流而中断的时间保证，您的生产代码也是如此。

标签： javascript rxjs rxjs5 reactivex

【解决方案1】：

下面的代码或许可以给你想要的结果，但没必要使用rxjs来操作数组恕我直言

Rx.Observable.combineLatest(
   Rx.Observable.from([0,0,0]),
   Rx.Observable.from([1,1,1])
).flatMap(value=>Rx.Observable.from(value))
.subscribe(console.log)

【讨论】：

抱歉，我没有足够清楚地说明我想要什么。我已对原始帖子进行了澄清并进行了编辑
.from 的默认行为不支持我对合并所做的测试中的异步，但适用于 zip 虽然 let a=Rx.Observable.from([1,1,1]) let b =Rx.Observable.from([2,2,2]) Rx.Observable.zip(a,b).subscribe(console.log)

【解决方案2】：

当流被分区时，不同子流中元素之间的时序保证实际上被破坏了。特别是，即使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 + 更复杂的累加器仍然是可行的。

【讨论】：