配置 Observable 以在推送新值时将所有先前的值作为数组返回？

Question

很难处理这个。我是使用 RxJs Observables 的新手，所以我需要一些指导。

我正在尝试构建一个可做两件事的 Observable 日志流。

1. 每当将新的行/值写入日志文件时，将该新值推送到流中。
2. 使用日志文件中的值开始预填充。

我已经满足了上述两个条件。现在的挑战是将它与 *ngFor 一起使用。

*ngFor 需要来自 Observable 的数组，因此它可以进行比较以添加/删除（我的最佳猜测）。但是我的 observable 只返回一个包含最后一个项目的数组。

//logviewer.page.ts constructor()
this.logs = Subject.create();
this.logs$ = this.logs.asObservable()
            .startWith("logs\\error.log")
            .flatMap((fileName: string) => {
                //start by reading the existing log files as a string
                return this.$localStorageService.readAsStringAsync(fileName);
            })
            .map((contents: string) => {
                //this part splits up the log file line-by-line into an log entry
                let logs = contents.split(/\r\n|\r|\n/).filter(n => n.length > 0);
                logs.forEach((s, ix, parent) => {
                    let x = JSON.parse(s);
                    parent[ix] = { timestamp: new Date(parseFloat(x[0])), message: x[1] };
                })
                return logs; //an array of objects { timestamp, message }
            })
            //merge the existing application log stream
            //throughout the application we log errors, info, etc
            //if a new entry is made it will appear here
            .merge(this.$loggerService.applicationLog$.map((x) => {                    
                //return an array with one object { timestamp, message }
                return [{ timestamp: new Date(parseFloat(x[0])), message: x[1] }];
            }))

现在我的模板非常简单。

//logviewer.template.ts
<div *ngFor="let entry of logs$ | async">
    {{entry|json}}
</div>

现在来测试一下，我有一个按钮来添加一个条目

//logviewer.page.ts
addEntry() {
    this.$loggerService.error("this is a test");
}

//LoggerService.service.ts
private applicationLog: ReplaySubject<any[]>;
get applicationLog$(): Observable<any[]> {
    return this.applicationLog.asObservable();
}

error(...args) {
    let data = [Date.now().toString()].concat(args.map<string>((n, ix) => { return toString(n); }));

    // ... write to file

    // fire the subject
    this.applicationLog.next(data);
}

现在，当我单击 addEntry 时，管道一切正常，并且通过可观察序列正确触发了该值。但是我的 *ngFor 只更新一个值。它不会保留所有先前日志条目的历史记录。只返回最后一个数组，这是有道理的。

如何让我的可观察序列总是返回一个包含所有值的数组。我可以让它一次返回一个条目，但我需要完整的历史来满足 *ngFor

我对 *ngFor 和异步管道缺乏了解。我以为它订阅了 observable，并自动将任何新条目添加到 ngFor，但事实并非如此。

Answer 1

尝试使用scan操作符：

this.logs$ = this.logs.asObservable()
        ...
        .merge(this.$loggerService.applicationLog$.map((x) => {                    
            //return an array with one object { timestamp, message }
            return [{ timestamp: new Date(parseFloat(x[0])), message: x[1] }];
        }))
        .scan((acc, x) => {
            acc.push(...x);
            return acc;
        }, []);