为每个订阅创建于扫描操作符的新种子对象

Question

RxJS 5.5.2

我有下面的代码谁分裂的数字阵列为对象，具有2个属性“小”的数字越小则4和'大'其余。

const o = from([1, 2, 3, 4, 5, 6]).pipe(
    scan<number, {}>((a, b) => { 
    if (b < 4) { 
     a['small'].push(b); 
    } else { 
     a['big'].push(b); 
    } 
    return a; 
    }, { 
    'small': [], 
    'big': [] 
    }) 
); 
console.log('subscription 1'); 
o.subscribe(x => console.log(JSON.stringify(x))); 
console.log('subscription 2'); 
o.subscribe(x => console.log(JSON.stringify(x))); 

认购1控制台打印后：

{"small":[1,2,3],"big":[4,5,6]} // this is ok 

订购2台打印后：

{"small":[1,2,3,1,2,3],"big":[4,5,6,4,5,6]} // this is not ok 

有没有开始一个新的种子对象每次有人赞同呢？

Answer 1

另一种选择是将管道封装在defer块中，该块将在订阅时重建源流。

defer(() => 
    from([1, 2, 3, 4, 5, 6]).pipe(
    scan<number, {}>((a, b) => { 
     if (b < 4) { 
     a['small'].push(b); 
     } else { 
     a['big'].push(b); 
     } 
     return a; 
    }, { 
     'small': [], 
     'big': [] 
    }) 
) 
); 

每个订阅都会调用推迟块中的方法并订阅结果。尽管像@arturgrzesiak提到的那样，变异数组在函数式编程和扩展功能反应式编程中被看作反模式。

Answer 2

扫描累加器（{ small: [], big: [] }）被.push突变，这是一种反模式，可能很容易导致意外的行为。防止改变先前发出的值

一种选择可能是：

scan<number, {}>((a, b) => { 
    if (b < 4) { 
    return Object.assign({}, a, {small: a.small.concat([b])}); 
    } else { 
    return Object.assign({}, a, {big: a.big.concat([b])}); 
    } 
}, { 
    'small': [], 
    'big': [] 
}) 

不知道你正在试图完成什么，但它可能是值得看一看的partition运营商，这将产生两个单独的值流如const [small, big] = someStream.partition(x => x < 4);。