RXJS扩展永远不会退出递归调用

Question

在上传文件到firebase后，我创建了一个递归调用来监听GCP触发日志，基本上工作得很好。我的问题是，我的递归扩展函数永远不会退出。正如你从代码中看到的，我已经检查了不止一次值，似乎一切都很好，我在控制台日志中也看到了ENTERED消息，但递归调用永远不会结束，没有任何错误。如果条件为true，如果我想中断或强制中断递归调用，我该怎么办？

public getLog(filePath: string): Observable<object[]> {
try {
...
return this.getLogChunk()
.pipe(
expand((data: any) => {
if (!environment.production && data && data.entries && data.entries.length > 0) {
console.groupCollapsed('GCP Service Info [getLog]');
console.info('[fileName]', fileName);
console.info('[some - Finished]', data.entries.some((x: any) => x.textPayload.includes('Finished')));
console.info('[some - Filename]', data.entries.some((x: any) => x.textPayload.includes(fileName)));
console.info('[some - Finished - Filename]', data.entries.some((x: any) => x.textPayload.includes('Finished') && x.textPayload.includes(fileName)));
console.info('[filter - Filename]', data.entries.filter((x: any) => x.textPayload.includes(fileName)));
console.groupEnd();
}
if (data &&
data.entries &&
data.entries.some((x: any) => x.textPayload.includes('Finished') && x.textPayload.includes(fileName))) {
console.log('ENTERED!!!!!');
return of({});
}
return this.getLogChunk().pipe(delay(2000));
}),
map(res => res),
reduce((acc: object[], val: any) => acc.concat(val), new Array<object>())
)
}
return new Observable<object[]>();
} catch (e) {
if (!environment.production) {
console.groupCollapsed('GCP Service Error [getLog]');
console.error('[error]', e);
console.groupEnd();
}
throw new Error(e);
}
}

在我调用getlog的地方，(你可以看到，我只检查了一次快照更改来触发getlog，当上传完成时，我对它进行了调试，没问题(，因此[complete] ENTERED从未触发，因为getlog中的无限递归循环。

...
const uploadTask: AngularFireUploadTask = this.storage.upload(filePath, fileToUpload);
...
uploadTask.snapshotChanges()
.pipe(
filter(task => {
if (task) {
return task.state === firebase.storage.TaskState.SUCCESS
}
return false;
})
)
.subscribe(val => {
this.gcpService.getLog(filePath)
.subscribe({
next: data => console.log('[next] ENTERED'),
complete: () => {
console.log('[complete] ENTERED');
}
});
})
...

Answer 1

我发现问题出在return of({})调用上——由于它提供了一个值，所以它也扩展了。如果希望扩展结束，请改为调用return EMPTY(从"Rxjs"导入(。EMTPY不返回值并立即完成，因此展开不会继续侦听。