RxJS - 如何使用toArray（）和异步observables数组？

Question

我使用Rx.Observable.create()创建了一个异步观察数组，并希望在完成时使用.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

该代码是正确的，除非您没有完成源observables。

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()似乎错在这里。在这段代码中，不需要使源观测数据变得很热。