如何在不完成添加序列的情况下从rxjs创建数组

我正试图找到一种rxjs方法来完成以下任务：

您有两个可观察性，一个是onAddObs，另一个是on RemoveObs。假设onAddObs.next（）激发几次，添加"a"、"B"、"C"。然后我想得到["A"，"B"，"C"]。

.toArray要求完成可观测。。。还有更多的事情可能发生。

这是第一部分。第二部分可能是显而易见的。。。我希望onRemoveObs然后从最终生成的数组中移除。

我没有钱，因为我做不到任何事情。。。提前感谢！

更新

根据用户3743222的建议，我签出了.scan，它完成了任务！如果其他人对此有问题，我已经提供了angular2服务，它展示了一种很好的方法。诀窍是使用.scan，而不是添加/删除的流，而是要添加/删除函数流，这样你就可以从扫描中调用它们并传递状态。

@Injectable() 
export class MyService {
    public items: Observable<any>;
    private operationStream: Subject<any>;
    constructor() {        
        this.operationStream = Subject.create();
        this.items = this.operationStream
            // This may look strange, but if you don't start with a function, scan will not run....so we seed it with an operation that does nothing.
            .startWith(items => items)
            // For every operation that comes down the line, invoke it and pass it the state, and get the new state. 
            .scan((state, operation:Function) => operation(state), [])
            .publishReplay(1).refCount();
        this.items.subscribe(x => {
            console.log('ITEMS CHANGED TO:', x);
        })    
    }
    public add(itemToAdd) {
        // create a function which takes state as param, returns new state with itemToAdd appended
        let fn = items => items.concat(itemToAdd);
        this.operationStream.next(fn);
    }
    public remove(itemToRemove) {
        // create a function which takes state as param, returns new array with itemToRemove filtered out 
        let fn = items => items.filter(item => item !== itemToRemove);
        this.operationStream.next(fn);
    }
}