如何为Observable添加外部complete处理器，确保自动释放数据源？

我有一个来自数据库查询接口的AsyncIterable，迭代完成后需要手动关闭。我需要基于该AsyncIterable创建一个Observable，使其能自定义complete逻辑。我知道可以在subscribe中处理complete，但我不想手动订阅，希望Observable能自动处理该逻辑。现有代码如下：

function fromQuery<T>(exec: QueryExecutor<T>): Observable<T> {
    const data: AsyncIterable<T> = exec.query();

    return from(data); // fails to release the source

    // I need to return an Observable that would always call
    // exec.release() inside its 'complete' handler.

    // how to do that?
}

如何完善fromQuery函数，让返回的Observable能确保任何订阅者在完成时都会调用exec.release()？

可以使用RxJS的finalize操作符，它会在Observable生命周期结束（正常完成、错误终止、订阅被取消）时执行指定的清理逻辑，刚好适合用来调用exec.release()。

修改后的代码如下：

import { from, finalize } from 'rxjs';

function fromQuery<T>(exec: QueryExecutor<T>): Observable<T> {
    const data: AsyncIterable<T> = exec.query();

    return from(data).pipe(
        finalize(() => {
            exec.release();
        })
    );
}

补充说明 #

• finalize的回调会覆盖所有订阅终止场景，不管是迭代正常完成、发生错误，还是订阅者主动取消订阅，都能确保exec.release()被调用，避免资源泄漏。
• 如果仅需在正常完成时调用释放逻辑，可以改用tap操作符的complete回调，但资源释放场景下finalize更安全，因为它不会遗漏异常或取消订阅的情况。

内容的提问来源于stack exchange，提问作者vitaly-t