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()似乎錯在這裏。在這段代碼中，不需要使源觀測數據變得很熱。