RxJS中from数组结合switchMap时finalize未触发问题排查

问题场景 #

我在NGXS状态类中编写了如下代码：

// code is in a state class for ngxs
const someObject = ctx.getState().someObject;
const sourceArray: SomeType[] = [/* content */];

return from(sourceArray).pipe(
  tap((arrayElem) => ctx.patchState({currentElem: arrayElem})),
  switchMap((arrayElem) => {
    return this.someService.someMethod(arrayElem);
  }),
  catchError((err) => {
    return of(err);
  }),
  finalize(() => {
    console.log("Test"); // Never called
  })
);


someMethod(request) {
    return this.someClientAsync().pipe(
      switchMap((theClient) => {
        return new Observable<ReturnType>(subscriber => {
            theClient.makeCall(request, (err, data) => {
            if (err) {
              subscriber.error(err)
            }
            subscriber.next(data);
          });
        });
      })
    );
  }

问题：switchMap能针对每个数组元素正常执行，但finalize方法从未被调用。我猜测问题出在someMethod中回调转Observable的实现上，虽然这种封装在服务其他方法里能正常工作，且switchMap确实执行了n次（n为数组长度）。

编辑补充
感谢大家的回复！我学到了很多之前不了解的知识。我确认问题确实出在someMethod中回调转Observable的实现方式上。我实际需要调用的API如下：

someClient.doAction(params, (data, err) => {
  // callback
});

我希望将其转换为Observable，目前打算采用@martin的方案，想确认是否存在更优的实现方式。

问题原因与解决方案 #

1. finalize未触发的根本原因 #

你封装回调API为Observable时，没有调用subscriber.complete()。finalize只有在Observable序列完成（调用complete）或者出错（调用error）时才会触发。你的代码里，成功场景下只调用了next但没有结束流，导致整个from(sourceArray)的流一直处于活跃状态，finalize自然不会执行。

2. 回调转Observable的最优实现方式 #

针对你实际的someClient.doAction回调API（参数顺序是data在前，err在后），有两种推荐的实现方式：

方式一：使用RxJS内置的bindCallback（推荐）

RxJS提供了bindCallback工具，可以直接将回调风格的函数转换为Observable，无需手动创建Observable：

import { bindCallback } from 'rxjs';

someMethod(request) {
  return this.someClientAsync().pipe(
    switchMap((theClient) => {
      // 绑定回调函数，指定结果选择器处理data和err的顺序
      const doAction$ = bindCallback((params, callback) => {
        theClient.doAction(params, callback);
      }, (data, err) => {
        if (err) throw err; // 抛出错误，让Observable捕获为error事件
        return data;
      });
      return doAction$(request);
    })
  );
}

bindCallback会自动处理流的完成逻辑：回调触发时，它会调用next然后立即complete，确保流正常结束，finalize就能被触发。

方式二：手动完善Observable封装

如果不想用bindCallback，可以手动在回调中调用complete：

someMethod(request) {
  return this.someClientAsync().pipe(
    switchMap((theClient) => {
      return new Observable<ReturnType>(subscriber => {
        theClient.doAction(request, (data, err) => {
          if (err) {
            subscriber.error(err);
            // error会自动终止流，无需额外调用complete
            return;
          }
          subscriber.next(data);
          subscriber.complete(); // 关键：成功后必须调用complete结束流
        });
      });
    })
  );
}

注意：调用subscriber.error(err)时，Observable会自动终止；但成功调用next后，必须手动调用complete来结束流。

3. 额外优化：数组遍历的并发控制 #

你当前用switchMap处理数组元素，会导致前一个请求被后一个请求取消（switchMap会切换到新的Observable并取消之前的）。如果需要调整执行策略：

• concatMap：按顺序依次执行，前一个完成后再执行下一个
• mergeMap：并行执行所有请求（可通过concurrency参数控制并发数）

根据业务需求选择即可，这属于额外优化点，和finalize触发问题无关。

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