RxJS 7.8.1中return setTimeout触发TypeError问题咨询

问题重现 #

以下代码在RxJS 7.8.1中运行会抛出错误：

import { Observable } from 'rxjs';

new Observable((sub) => {
    return setTimeout(() => {
        sub.next("Test");
        sub.complete();
    }, 1000);
}).subscribe({
    next(value) {
        console.log(value);
    }
});

错误输出：

TypeError: finalizer.unsubscribe is not a function
    at execFinalizer (/home/jsturm/Alfred/template-service/node_modules/rxjs/dist/cjs/internal/Subscription.js:175:19)
    at Subscription.unsubscribe (/home/jsturm/Alfred/template-service/node_modules/rxjs/dist/cjs/internal/Subscription.js:89:29)
    at Subscriber.unsubscribe (/home/jsturm/Alfred/template-service/node_modules/rxjs/dist/cjs/internal/Subscriber.js:75:42)
    at Subscriber._complete (/home/jsturm/Alfred/template-service/node_modules/rxjs/dist/cjs/internal/Subscriber.js:95:18)
    at Subscriber.complete (/home/jsturm/Alfred/template-service/node_modules/rxjs/dist/cjs/internal/Subscriber.js:69:18)
    at Timeout._onTimeout (file:///home/jsturm/Alfred/template-service/test/observable.js:7:13)
    at listOnTimeout (node:internal/timers:569:17)
    at process.processTimers (node:internal/timers:512:7)

移除return语句后代码可正常运行，且该问题在RxJS 6.x版本中不会出现。

错误原因 #

1. RxJS 7对订阅函数返回值的严格要求：Observable构造函数的订阅回调如果有返回值，必须是以下两种类型之一：
   • 无参数的清理函数：RxJS会自动将其包装为带有unsubscribe方法的Subscription对象，用于订阅取消时执行清理；
   • 带有unsubscribe方法的对象（比如Subscription实例）。
2. setTimeout返回值不符合要求：setTimeout返回的是数字类型的定时器ID，既不是函数，也没有unsubscribe方法。RxJS 7会将这个返回值当作清理终结器，尝试调用其unsubscribe方法，因此抛出类型错误。
3. RxJS 6的差异：RxJS 6对订阅函数返回值的处理更宽松，若返回值不是符合要求的清理对象/函数，会直接忽略，不会尝试调用unsubscribe，因此不会触发错误。

修复方案 #

方案1：无需清理逻辑时，移除return语句 #

如果不需要在订阅提前取消时终止定时器（比如定时器一定会执行到complete），直接去掉return即可：

import { Observable } from 'rxjs';

new Observable((sub) => {
    setTimeout(() => {
        sub.next("Test");
        sub.complete();
    }, 1000);
}).subscribe({
    next(value) {
        console.log(value);
    }
});

方案2：需要清理逻辑时，返回清理函数 #

如果希望在订阅提前取消时清理定时器，返回一个清除定时器的函数即可：

import { Observable } from 'rxjs';

new Observable((sub) => {
    const timerId = setTimeout(() => {
        sub.next("Test");
        sub.complete();
    }, 1000);
    // 订阅取消时会调用此函数清除定时器
    return () => clearTimeout(timerId);
}).subscribe({
    next(value) {
        console.log(value);
    }
});

关于错误提示的建议 #

当前错误提示仅指出finalizer.unsubscribe is not a function，无法直接定位到“订阅函数返回值类型不符合要求”的核心问题。建议RxJS在检测到订阅函数返回值既不是函数也不是带有unsubscribe方法的对象时，抛出更明确的错误信息，比如：

Observable订阅函数的返回值必须是无参数函数或具有unsubscribe方法的对象，当前返回的是[具体类型]。

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