RxJS：异步执行器创建Observable的类型问题与标准方案

问题描述 #

我想实现以下异步数据获取逻辑，但受限于Observable构造器的类型约束无法直接完成：

const fetchData = async (observer: Observer<ViewState<...>>): Promise<void> => {
  observer.next(loading())

  const response = await ... // 发起数据请求

  if (response.type === ResponseType.FAILURE) {
    observer.next(fail())
  } else {
    observer.next(loaded(response.data))
  }
}

// 此处报错：Observable构造器要求执行器返回`TeardownLogic`类型，而非`Promise<T>`
const obs = new Observable(fetchData)
const subscription = obs.subscribe(...)

核心矛盾：Observable构造器的执行函数必须返回TeardownLogic（清理逻辑，可以是函数、Subscription或void），不能直接返回Promise。

我当前的实现 #

为了绕开这个限制，我自行编写了一个辅助构建函数：

export const buildObservableFromAsyncExecutor = <T>(
  executor: (subscriber: Observer<T>) => Promise<void>
): Observable<T> => {
  const subject = new Subject<T>()
  executor(subject) // 未等待Promise完成

  return subject.asObservable()
}

// 使用示例
const obs = buildObservableFromAsyncExecutor(fetchData)
const subscription = obs.subscribe(...)

规范实现方式 #

方式1：直接在Observable构造器内处理异步逻辑 #

这是最符合RxJS设计规范的写法，直接在构造器的执行函数中封装异步逻辑，同时返回合法的清理逻辑：

const obs = new Observable<ViewState<...>>(async (observer) => {
  // 发出加载中状态
  observer.next(loading())

  try {
    const response = await ... // 发起数据请求
    // 根据响应结果发出对应状态
    if (response.type === ResponseType.FAILURE) {
      observer.next(fail())
    } else {
      observer.next(loaded(response.data))
    }
    // 可选：通知数据流结束
    observer.complete()
  } catch (error) {
    // 捕获异步错误并通知观察者
    observer.error(error)
  }

  // 返回清理逻辑（比如取消未完成的请求）
  return () => {
    // 示例：如果使用AbortController，这里可以调用controller.abort()
  }
})

const subscription = obs.subscribe(...)

这种写法的优势：

• 完全遵循RxJS的设计范式，无需额外封装
• 能捕获异步过程中的错误，避免未捕获Promise异常
• 支持订阅取消时的资源清理，防止内存泄漏

方式2：使用RxJS内置defer操作符 #

如果希望保留异步逻辑的独立性，可以用defer操作符包装，它会在每次订阅时执行传入的异步函数：

写法一：结合startWith发出初始状态

const fetchData = async (): Promise<ViewState<...>> => {
  const response = await ... // 发起数据请求
  return response.type === ResponseType.FAILURE ? fail() : loaded(response.data)
}

// 用startWith发出初始loading状态，defer执行异步请求
const obs = defer(() => fetchData()).pipe(
  startWith(loading())
)

const subscription = obs.subscribe(...)

写法二：在defer内直接处理完整逻辑

const obs = defer(async () => {
  const response = await ... // 发起数据请求
  return response.type === ResponseType.FAILURE ? fail() : loaded(response.data)
}).pipe(
  startWith(loading())
)

优化你的辅助函数 #

你当前的辅助函数存在两个隐患：未处理异步错误、缺少清理逻辑。可以修改为符合规范的版本：

export const buildObservableFromAsyncExecutor = <T>(
  executor: (subscriber: Observer<T>) => Promise<void | (() => void)>
): Observable<T> => {
  return new Observable<T>(async (observer) => {
    let teardown: (() => void) | undefined;
    try {
      // 等待异步执行器完成，并获取可能的清理函数
      teardown = await executor(observer);
    } catch (error) {
      // 将异步错误转发给观察者
      observer.error(error);
    }
    // 返回组合后的清理逻辑
    return () => {
      teardown?.();
    };
  });
}

这个版本既满足Observable构造器的类型要求，又能处理异步错误，还支持资源清理。

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