如何在RXJS中等待Subject.next后的异步操作完成？

我正在对一个Angular组件进行单元测试，该组件依赖的服务提供了Observable。组件在ngOnInit中订阅该Observable，回调内执行异步方法doSomethingWithValue，该方法会调用服务的异步方法doSomeAsyncCall。

组件与服务代码 #

export class SomeComponent implements OnInit{
    private _subscription:Subscription;

    constructor(private someService:SomeService){
    }
    
    ngOnInit(){
        this.subscription = this.someService.someObservable.subscribe(async v=>await this.doSomethingWithValue(v));
    }
    
    ngOnDestroy(){
        this.subscription.unsubscribe();
    }
    
    async doSomethingWithValue(v){
        await this.someService.doSomeAsyncCall();
        //...   
    }   
}

export class SomeService{
    public someObservable:Observable<string>;
    
    constructor(){
        this.someObservable = //subscription to multiple other services, transformations and consolidation.
    }
    
    async doSomeAsyncCall(){
        await something...
    }
}

测试代码（使用ng-mocks） #

it('should update correctly when XYZ', async ()=> {
    
    const notifications = new Subject<string>();
    MockInstance(SomeService, 'someObservable', notifications.asObservable());  
    const component = MockRender(SomeComponent).point.componentInstance;
    
    notifications.next('something');
    
    expect(something);//This is being called before the `doSomethingWithValue` get called
    
})

问题核心：调用notifications.next('something')后，expect语句会在doSomethingWithValue的异步操作完成前执行。由于notifications.next()不返回Observable或Promise，无法直接等待。需要确保Subject触发后所有异步调用都完成，且不能用随机等待时间，该问题也可能出现在业务代码中。

一、测试场景下的处理 #

方法1：捕获异步回调的Promise #

由于订阅的async回调本质返回Promise，我们可以在测试中捕获这个Promise并等待它完成：

it('should update correctly when XYZ', async ()=> {
    const notifications = new Subject<string>();
    let callbackPromise: Promise<void>;
    
    // 拦截Observable，捕获异步回调的Promise
    MockInstance(SomeService, 'someObservable', () => {
        return notifications.asObservable().pipe(
            tap(v => {
                callbackPromise = component.doSomethingWithValue(v);
            })
        );
    });
    
    const fixture = MockRender(SomeComponent);
    const component = fixture.point.componentInstance;
    
    notifications.next('something');
    // 等待异步操作完成
    await callbackPromise;
    
    expect(something).toBe(true);
})

方法2：使用Angular的fixture.whenStable() #

Angular测试中的whenStable()会等待所有待处理的异步操作（包括Promise）完成：

it('should update correctly when XYZ', async ()=> {
    const notifications = new Subject<string>();
    MockInstance(SomeService, 'someObservable', notifications.asObservable());  
    const fixture = MockRender(SomeComponent);
    const component = fixture.point.componentInstance;
    
    notifications.next('something');
    // 等待所有异步任务完成
    await fixture.whenStable();
    
    expect(something).toBe(true);
})

注意：如果doSomeAsyncCall是模拟方法，需确保它返回的Promise能被正确resolve，否则whenStable()无法感知。

方法3：等待微任务队列清空 #

如果异步操作都在微任务队列中，可通过await Promise.resolve()等待队列清空：

it('should update correctly when XYZ', async ()=> {
    const notifications = new Subject<string>();
    // 模拟异步方法为已resolve的Promise
    MockInstance(SomeService, 'doSomeAsyncCall', async () => Promise.resolve());
    
    MockInstance(SomeService, 'someObservable', notifications.asObservable());  
    const component = MockRender(SomeComponent).point.componentInstance;
    
    notifications.next('something');
    // 等待微任务队列清空
    await Promise.resolve();
    
    expect(something).toBe(true);
})

二、业务场景下的处理 #

方法1：将异步逻辑转为Observable流 #

避免在subscribe回调中使用async/await，改用RxJS操作符将异步逻辑转为Observable流，方便后续等待：

// 修改组件的ngOnInit
ngOnInit(){
    this.subscription = this.someService.someObservable.pipe(
        // 用switchMap把Promise转为Observable
        switchMap(v => from(this.doSomethingWithValue(v)))
    ).subscribe();
}

业务代码中需要等待时，可通过Observable操作符实现：

// 等待流完成
await this.someService.someObservable.pipe(
    switchMap(v => from(this.doSomethingWithValue(v))),
    first()
).toPromise();

方法2：维护异步任务队列 #

在组件中维护一个Promise队列，提供方法供外部等待所有异步任务完成：

export class SomeComponent implements OnInit{
    private _subscription:Subscription;
    private asyncTasks: Promise<void>[] = [];

    constructor(private someService:SomeService){
    }
    
    ngOnInit(){
        this.subscription = this.someService.someObservable.subscribe(v=> {
            const task = this.doSomethingWithValue(v);
            this.asyncTasks.push(task);
            // 任务完成后从队列移除
            task.finally(() => {
                const index = this.asyncTasks.indexOf(task);
                if (index !== -1) this.asyncTasks.splice(index, 1);
            });
        });
    }
    
    // 供外部调用，等待所有异步任务完成
    async waitForAllTasks(): Promise<void> {
        await Promise.all([...this.asyncTasks]);
    }
    
    // ...其他代码
}

业务代码中可直接调用component.waitForAllTasks()等待所有异步操作完成。

内容的提问来源于stack exchange，提问作者J4N