fromArray Rx wiki on github

coffee> rext = require 'rx'                                                 
coffee> arr = [1..5]                                                 
[ 1, 2, 3, 4, 5 ]                                                    
coffee> obs = rext.Observable.fromArray(arr)                         
{ _subscribe: [Function] }                                           
coffee> obs.subscribe( (x) -> console.log("added value: " + x))      
added value: 1                                                       
added value: 2                                                       
added value: 3                                                       
added value: 4                                                       
added value: 5                                                       
{ isStopped: true,                                                   
  observer:                                                          
   { isStopped: true,                                                
     _onNext: [Function],                                            
     _onError: [Function: defaultError],                             
     _onCompleted: [Function: noop] },                               
  m: { isDisposed: true, current: null } }                           
coffee> arr.push(12)    # expecting "added value: 12"                                              
6                       # instead got new length of array                                              
coffee>          

它看起来像订阅函数只会在创建时触发一次.看起来它有点用词不当,因为我真的只是为了 – 每个数组而不是观察它的变化.该代码几乎与维基上的代码完全相同.所以要么我做错了,要么订阅不符合我的预期.

解决方法:

当您添加订阅服务器时,Observable.fromArray会创建一个Observable,它会立即触发每个数组项的事件.因此,它不会“观察”该阵列的更改.

如果你需要一个“可推送的集合”,Bacon.js中的Bus类可能就是你想要的.对于RxJs,我的小MessageQueue类具有类似的功能.