rxjs - 如何重置 ReplaySubject 上的扫描累加器？

Question

我有一个 ReplaySubject，它使用 scan 运算符累积数据，每 10000 毫秒应重置一次。还有其他方法吗？

现在:

let subject = new ReplaySubject();

subject.scan((acc, cur) => {
            acc.push(cur);
            return acc;
        }, [])
        .subscribe(events => {
            localStorage.setItem('data', JSON.stringify(events))
        });

subject
    .bufferTime(10000)
    .map(() => {
        subject.observers[0]._seed = [];
    })
    .subscribe(() => localStorage.removeItem('data'));

Answer 1

我前几天问过一个非常相似的问题，后来自己回答了

accumulating values such as with scan but with the possibility to reset the accumulator over time

也许这对你有帮助

更多细节

另一种方法是让 Observable 充当计时器，它以固定的时间间隔发出，在您的情况下为 10000 毫秒。

一旦这个计时器 发出，您将控制传递给通过scan 运算符累积的Observable。要传递控制权，您可以使用 switchMap 运算符来确保 Observable 的前一个实例完成。

如果我正确理解你想要实现的目标，我会使用普通的 Subject 而不是 ReplaySubject。

代码可能是这样的

const subject = new Subject<number>();

const timer = Observable.timer(0, 1000).take(4);

const obs = timer.switchMap(
    () => {
        console.log('-----');
        return subject
            .scan((acc, cur) => {
                acc.push(cur);
                return acc;
            }, []);
    }
)

obs.subscribe(
    events => {
        console.log(JSON.stringify(events))
    }
);


// TEST DATA EMITTED BY THE SUBJECT
setTimeout(() => {
    subject.next(1);
}, 100);
setTimeout(() => {
    subject.next(2);
}, 1100);
setTimeout(() => {
    subject.next(3);
}, 2100);
setTimeout(() => {
    subject.next(4);
}, 2200);