reactive-programming - RxJS - 如何将 toArray() 与异步 observables 数组一起使用？

Question

我正在使用 Rx.Observable.create() 创建一个异步 observables 数组并希望使用 .toArray()完成后获取所有值。

console.log('running');
let valsArray = ['a','b','c'].map((val,i)=>{
  return Rx.Observable.create((obs)=>{
    let tid = setTimeout(()=>{
      console.log(val + ' timing out');
      obs.onNext(val);
    },i*500);
    return ()=>{
      clearTimeout(tid);
    };
  }).publish().refCount();
});

Rx.Observable.from(valsArray)
.flatMap((v)=>v)
.toArray()
.subscribe((arr)=>{
  console.log("arr should be ['a','b','c']",arr);
});

以上示例位于 http://jsbin.com/wegoha/10/edit?js,console .

使用 setTimeout作为其他异步操作的替代品，以保持示例简单。

Answer 1

代码是正确的，除非您没有完成源可观察对象。
toArray()操作符只能在 observable 完成时工作，而且由于您没有完成 Rx.Observable.create那么您的查询永远不会结束。

尝试这个:

console.log('running');
let valsArray = ['a','b','c'].map((val,i)=>{
  return Rx.Observable.create((obs)=>{
    let tid = setTimeout(()=>{
      console.log(val + ' timing out');
      obs.onNext(val);
      obs.onCompleted();
    },i*500);
    return ()=>{
      clearTimeout(tid);
    };
  }).publish().refCount();
});

Rx.Observable.from(valsArray)
.flatMap((v)=>v)
.toArray()
.subscribe((arr)=>{
  console.log("arr should be ['a','b','c']",arr);
});

另外，作为旁注， .publish().refCount()这里似乎错了。在这段代码中没有必要使源 observables 变热。