后端【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址https://gitcode.com/gh_mirrors/rxj/RxJS点击查看免费下载本文是 RxJS v4Reactive Extensions for JavaScript入门系列的一部分聚焦于如何为 RxJS 编写自定义的 Observable 操作符。你将掌握两条扩展路径使用Rx.Observable.create从零实现操作符以及通过组合filter、map、mergeAll等内置操作符快速构建语义化的新操作符同时学会用TestScheduler与collectionAssert为自定义操作符编写可回归的单元测试。读完本文你可以在不修改库源码的前提下为项目注入可复用、可测试、语义清晰的专属操作符。为什么要实现自己的操作符RxJS 提供了一套相当完整的内置操作符覆盖了对数据集合的大多数常见操作筛选、投影、合并、聚合、时间控制等。但在真实项目中你仍可能遇到两种需要扩展的场景补充缺失的语义内置操作符无法直接表达你的领域语义而该语义可能在代码中反复出现值得封装成一个可复用的操作符。封装与可读性把一串职责相近的内置操作符组合包装成一个名字更有意义的操作符让查询意图一目了然。例如Lo-Dash 与 Underscore 提供了_.where方法传入一组属性对集合元素做深比较deep equality筛选出属性匹配的元素。RxJS 内置的filter只接受一个谓词函数并不直接支持按属性集合匹配这时就可以把它封装成一个自定义操作符filterByProperties。方案一用Rx.Observable.create从零实现最直接的做法是使用Rx.Observable.create手动实现订阅逻辑。在源码中Rx.Observable.create实际返回一个AnonymousObservable// src/core/linq/observable/create.js Observable.create function (subscribe, parent) { return new AnonymousObservable(subscribe, parent); };也就是说create接收一个订阅函数实现并把它包装成可观测序列。基于此我们可以实现filterByPropertiesRx.Observable.prototype.filterByProperties function (properties) { var source this, comparer Rx.internals.isEqual; return Rx.Observable.create(function (observer) { // Our disposable is the subscription from the parent return source.subscribe( function (data) { try { var shouldRun true; // Iterate the properties for deep equality for (var prop in properties) { if (!comparer(properties[prop], data[prop])) { shouldRun false; break; } } } catch (e) { observer.onError(e); } if (shouldRun) { observer.onNext(data); } }, observer.onError.bind(observer), observer.onCompleted.bind(observer) ); }); };这段代码的核心要点订阅转发create内部函数返回的 disposable 就是父序列的订阅实现了背压链路上的资源传递。深比较循环遍历properties的每个属性调用Rx.internals.isEqual做深比较任何一个属性不匹配即判定为不通过。错误处理比较过程包在try/catch中异常通过observer.onError(e)传播而不是静默吞掉。通知转发onError与onCompleted直接绑定到上游 observer保证错误与完成信号正确向下游传递。其中Rx.internals.isEqual并非简单的而是实现了完整的深度相等比较。查看 src/core/internal/isequal.js 可以看到它支持数组、普通对象、Date、RegExp、Error、Map/Set、TypedArray 等类型的递归比较并且用stackA/stackB处理了循环引用场景因此可以放心用于嵌套结构的匹配。方案二组合已有操作符实现filterByProperties的逻辑本质就是按谓词过滤而 RxJS 内置的filter别名where正好承担这一职责。源码中filter的定义如下// src/core/perf/operators/filter.js observableProto.filter observableProto.where function (predicate, thisArg) { // ... };因此我们可以用filter重写filterByProperties把深比较封装进谓词Rx.Observable.prototype.filterByProperties function (properties) { var comparer Rx.internals.isEqual; return this.filter(function (data) { // Iterate the properties for deep equality for (var prop in properties) { if (!comparer(properties[prop], data[prop])) { return false; } } return true; }); };与从零实现相比代码量更少且直接继承filter内置的性能优化与异常处理能力。这种用操作符组合操作符正是 RxJS 内部的一贯做法。例如flatMap别名selectMany、mergeMap并不是凭空实现的而是由投影与扁平化合并组合而成。在 src/core/perf/operators/flatmap.js 中可以看到observableProto.flatMap observableProto.selectMany observableProto.mergeMap function (selector, resultSelector, thisArg) { return new FlatMapObservable(this, selector, resultSelector, thisArg).mergeAll(); };FlatMapObservable负责对每个元素应用selectormergeAll定义于 src/core/perf/operators/mergeall.js负责把产生的一层层内部序列扁平化合并。类似地buffer、bufferWithTime等操作符内部也复用了windowWithTimeflatMap的组合见 src/core/linq/observable/bufferwithtime.js。这也解释了为什么文档中flatMap可以如此简洁地写成Rx.Observable.prototype.flatMap function (selector) { return this.map(selector).mergeObservable(); };在本文所基于的 v4 仓库中该组合已被优化为FlatMapObservablemergeAll()的实现但组合思路一致。两种方案的取舍与资源管理规范维度Rx.Observable.create从零实现组合已有操作符灵活性最高可完全控制订阅与通知逻辑受限于现有操作符的语义代码量较大需自行处理错误与转发较小更易读性能/异常处理需自行实现直接继承内置实现适用场景内置操作符无法表达的自定义行为标准查询逻辑的语义化封装文档特别强调一个最佳实践编写自定义操作符时不要遗留任何未使用的 disposable。create的订阅函数应返回并透传父订阅如方案一中的return source.subscribe(...)否则可能出现资源泄漏并且取消订阅dispose无法正确沿链路传播。另一个实践要点是优先考虑用现有操作符组合实现。当你的自定义行为本质上等价于某个内置操作符如按属性过滤本质是过滤组合方案能以更少代码获得与库同等水平的健壮性。测试你的自定义操作符写完实现并不等于结束。RxJS 提供了TestScheduler虚拟时间调度器来测试这类序列无需真实等待时间流逝。下面为filterByProperties编写测试测试基础设施collectionAssert.assertEqual来自 测试与调试指南var onNext Rx.ReactiveTest.onNext, onCompleted Rx.ReactiveTest.onCompleted, subscribe Rx.ReactiveTest.subscribe; test(filterProperties should yield with match, function () { var scheduler new Rx.TestScheduler(); var input scheduler.createHotObservable( onNext(210, { name: curly, age: 30, quotes: [Oh, a wise guy, eh?, Poifect!] }), onNext(220, { name: moe, age: 40, quotes: [Spread out!, You knucklehead!] }), onCompleted(230) ); var results scheduler.startWithCreate( function () { return input.filterByProperties({ age: 40 }); } ); collectionAssert.assertEqual(results.messages, [ onNext(220, { name: moe, age: 40, quotes: [Spread out!, You knucklehead!] }), onCompleted(230) ]); collectionAssert.assertEqual(input.subscriptions, [ subscribe(200, 230) ]); });这段测试的要点createHotObservable创建在虚拟时间轴 210/220/230 上发布数据的热序列startWithCreate在虚拟时间 200 订阅、序列结束后自动处理返回记录全部通知的results.messages断言消息collectionAssert.assertEqual比较实际收到的通知序列——只有age: 40的moe被放行且onCompleted在 230 准时触发断言订阅input.subscriptions记录了subscribe(200, 230)验证订阅在预期的时间区间内发生并正确解除。类似的测试模式在仓库中有大量真实用例可参考例如 tests/observable/where.js 对内置filter的测试它用createHotObservable构造带精确时间戳的输入序列对完整过滤空输入异常传播等场景逐一断言results.messages与xs.subscriptions这套方法论完全可以平移到自定义操作符上。测试覆盖的边界情况要让测试真正可靠官方建议至少覆盖以下场景无匹配数据所有元素都被过滤只产生onCompleted空序列输入为空时行为正确单条匹配恰好一条数据通过多条匹配多条数据通过且顺序保持错误传播属性比较抛出异常时onError是否按预期触发订阅生命周期订阅是否在正确的时间点建立与解除对应input.subscriptions断言。由于测试基于虚拟时间全部断言在毫秒级完成非常适合集成到 CI 或常规测试任务中持续回归。小结自定义操作符是扩展 RxJS 能力的标准方式核心方法论可以归纳为三点优先组合其次从零能用filter、map、mergeAll等内置操作符组合出语义时直接组合享受内置的性能与异常处理确实需要底层控制时再用Rx.Observable.create。管理好 disposable始终透传父订阅避免资源泄漏与取消失效。用TestScheduler驱动开发通过createHotObservablestartWithCreatecollectionAssert.assertEqual验证通知序列与订阅区间并覆盖空、单匹配、多匹配、异常等边界。延伸阅读测试与调试 RxJS 应用本文测试基础设施的完整来源包含collectionAssert实现、do调试与长堆栈支持创建与订阅简单 Observable 序列查询 Observable 序列按类别浏览操作符操作符 API 参考赞分享后端【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址https://gitcode.com/gh_mirrors/rxj/RxJS点击查看免费下载相关推荐RxJS自定义操作符测试TDD开发流程RxJS自定义操作符测试TDD开发流程 你是否在开发RxJS操作符时遇到过难以调试的边界情况是否希望每一次代码变更都能安全可靠本文将通过TDD测试驱动开前端Javacord实战项目从零构建功能完整的Discord管理机器人Javacord实战项目从零构建功能完整的Discord管理机器人 你是否想拥有一个属于自己的Discord管理机器人却苦于找不到简单易用的Java开发库RxJava 自定义 Observable 操作符完全指南lift 序列操作符与 compose 转换操作符的源码级实战RxJava 自定义 Observable 操作符完全指南lift 序列操作符与 compose 转换操作符的源码级实战 在 RxJava 中编写自定义 Ob后端异步编程上一篇AssetRipper 快速上手指南从 Unity 游戏里完整提取资源下一篇革命性电话号码处理工具libphonenumber彻底解决全球号码格式混乱难题创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
RxJS 自定义 Observable 操作符实战：从 `Rx.Observable.create` 到组合已有操作符与测试驱动开发
后端【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址https://gitcode.com/gh_mirrors/rxj/RxJS点击查看免费下载本文是 RxJS v4Reactive Extensions for JavaScript入门系列的一部分聚焦于如何为 RxJS 编写自定义的 Observable 操作符。你将掌握两条扩展路径使用Rx.Observable.create从零实现操作符以及通过组合filter、map、mergeAll等内置操作符快速构建语义化的新操作符同时学会用TestScheduler与collectionAssert为自定义操作符编写可回归的单元测试。读完本文你可以在不修改库源码的前提下为项目注入可复用、可测试、语义清晰的专属操作符。为什么要实现自己的操作符RxJS 提供了一套相当完整的内置操作符覆盖了对数据集合的大多数常见操作筛选、投影、合并、聚合、时间控制等。但在真实项目中你仍可能遇到两种需要扩展的场景补充缺失的语义内置操作符无法直接表达你的领域语义而该语义可能在代码中反复出现值得封装成一个可复用的操作符。封装与可读性把一串职责相近的内置操作符组合包装成一个名字更有意义的操作符让查询意图一目了然。例如Lo-Dash 与 Underscore 提供了_.where方法传入一组属性对集合元素做深比较deep equality筛选出属性匹配的元素。RxJS 内置的filter只接受一个谓词函数并不直接支持按属性集合匹配这时就可以把它封装成一个自定义操作符filterByProperties。方案一用Rx.Observable.create从零实现最直接的做法是使用Rx.Observable.create手动实现订阅逻辑。在源码中Rx.Observable.create实际返回一个AnonymousObservable// src/core/linq/observable/create.js Observable.create function (subscribe, parent) { return new AnonymousObservable(subscribe, parent); };也就是说create接收一个订阅函数实现并把它包装成可观测序列。基于此我们可以实现filterByPropertiesRx.Observable.prototype.filterByProperties function (properties) { var source this, comparer Rx.internals.isEqual; return Rx.Observable.create(function (observer) { // Our disposable is the subscription from the parent return source.subscribe( function (data) { try { var shouldRun true; // Iterate the properties for deep equality for (var prop in properties) { if (!comparer(properties[prop], data[prop])) { shouldRun false; break; } } } catch (e) { observer.onError(e); } if (shouldRun) { observer.onNext(data); } }, observer.onError.bind(observer), observer.onCompleted.bind(observer) ); }); };这段代码的核心要点订阅转发create内部函数返回的 disposable 就是父序列的订阅实现了背压链路上的资源传递。深比较循环遍历properties的每个属性调用Rx.internals.isEqual做深比较任何一个属性不匹配即判定为不通过。错误处理比较过程包在try/catch中异常通过observer.onError(e)传播而不是静默吞掉。通知转发onError与onCompleted直接绑定到上游 observer保证错误与完成信号正确向下游传递。其中Rx.internals.isEqual并非简单的而是实现了完整的深度相等比较。查看 src/core/internal/isequal.js 可以看到它支持数组、普通对象、Date、RegExp、Error、Map/Set、TypedArray 等类型的递归比较并且用stackA/stackB处理了循环引用场景因此可以放心用于嵌套结构的匹配。方案二组合已有操作符实现filterByProperties的逻辑本质就是按谓词过滤而 RxJS 内置的filter别名where正好承担这一职责。源码中filter的定义如下// src/core/perf/operators/filter.js observableProto.filter observableProto.where function (predicate, thisArg) { // ... };因此我们可以用filter重写filterByProperties把深比较封装进谓词Rx.Observable.prototype.filterByProperties function (properties) { var comparer Rx.internals.isEqual; return this.filter(function (data) { // Iterate the properties for deep equality for (var prop in properties) { if (!comparer(properties[prop], data[prop])) { return false; } } return true; }); };与从零实现相比代码量更少且直接继承filter内置的性能优化与异常处理能力。这种用操作符组合操作符正是 RxJS 内部的一贯做法。例如flatMap别名selectMany、mergeMap并不是凭空实现的而是由投影与扁平化合并组合而成。在 src/core/perf/operators/flatmap.js 中可以看到observableProto.flatMap observableProto.selectMany observableProto.mergeMap function (selector, resultSelector, thisArg) { return new FlatMapObservable(this, selector, resultSelector, thisArg).mergeAll(); };FlatMapObservable负责对每个元素应用selectormergeAll定义于 src/core/perf/operators/mergeall.js负责把产生的一层层内部序列扁平化合并。类似地buffer、bufferWithTime等操作符内部也复用了windowWithTimeflatMap的组合见 src/core/linq/observable/bufferwithtime.js。这也解释了为什么文档中flatMap可以如此简洁地写成Rx.Observable.prototype.flatMap function (selector) { return this.map(selector).mergeObservable(); };在本文所基于的 v4 仓库中该组合已被优化为FlatMapObservablemergeAll()的实现但组合思路一致。两种方案的取舍与资源管理规范维度Rx.Observable.create从零实现组合已有操作符灵活性最高可完全控制订阅与通知逻辑受限于现有操作符的语义代码量较大需自行处理错误与转发较小更易读性能/异常处理需自行实现直接继承内置实现适用场景内置操作符无法表达的自定义行为标准查询逻辑的语义化封装文档特别强调一个最佳实践编写自定义操作符时不要遗留任何未使用的 disposable。create的订阅函数应返回并透传父订阅如方案一中的return source.subscribe(...)否则可能出现资源泄漏并且取消订阅dispose无法正确沿链路传播。另一个实践要点是优先考虑用现有操作符组合实现。当你的自定义行为本质上等价于某个内置操作符如按属性过滤本质是过滤组合方案能以更少代码获得与库同等水平的健壮性。测试你的自定义操作符写完实现并不等于结束。RxJS 提供了TestScheduler虚拟时间调度器来测试这类序列无需真实等待时间流逝。下面为filterByProperties编写测试测试基础设施collectionAssert.assertEqual来自 测试与调试指南var onNext Rx.ReactiveTest.onNext, onCompleted Rx.ReactiveTest.onCompleted, subscribe Rx.ReactiveTest.subscribe; test(filterProperties should yield with match, function () { var scheduler new Rx.TestScheduler(); var input scheduler.createHotObservable( onNext(210, { name: curly, age: 30, quotes: [Oh, a wise guy, eh?, Poifect!] }), onNext(220, { name: moe, age: 40, quotes: [Spread out!, You knucklehead!] }), onCompleted(230) ); var results scheduler.startWithCreate( function () { return input.filterByProperties({ age: 40 }); } ); collectionAssert.assertEqual(results.messages, [ onNext(220, { name: moe, age: 40, quotes: [Spread out!, You knucklehead!] }), onCompleted(230) ]); collectionAssert.assertEqual(input.subscriptions, [ subscribe(200, 230) ]); });这段测试的要点createHotObservable创建在虚拟时间轴 210/220/230 上发布数据的热序列startWithCreate在虚拟时间 200 订阅、序列结束后自动处理返回记录全部通知的results.messages断言消息collectionAssert.assertEqual比较实际收到的通知序列——只有age: 40的moe被放行且onCompleted在 230 准时触发断言订阅input.subscriptions记录了subscribe(200, 230)验证订阅在预期的时间区间内发生并正确解除。类似的测试模式在仓库中有大量真实用例可参考例如 tests/observable/where.js 对内置filter的测试它用createHotObservable构造带精确时间戳的输入序列对完整过滤空输入异常传播等场景逐一断言results.messages与xs.subscriptions这套方法论完全可以平移到自定义操作符上。测试覆盖的边界情况要让测试真正可靠官方建议至少覆盖以下场景无匹配数据所有元素都被过滤只产生onCompleted空序列输入为空时行为正确单条匹配恰好一条数据通过多条匹配多条数据通过且顺序保持错误传播属性比较抛出异常时onError是否按预期触发订阅生命周期订阅是否在正确的时间点建立与解除对应input.subscriptions断言。由于测试基于虚拟时间全部断言在毫秒级完成非常适合集成到 CI 或常规测试任务中持续回归。小结自定义操作符是扩展 RxJS 能力的标准方式核心方法论可以归纳为三点优先组合其次从零能用filter、map、mergeAll等内置操作符组合出语义时直接组合享受内置的性能与异常处理确实需要底层控制时再用Rx.Observable.create。管理好 disposable始终透传父订阅避免资源泄漏与取消失效。用TestScheduler驱动开发通过createHotObservablestartWithCreatecollectionAssert.assertEqual验证通知序列与订阅区间并覆盖空、单匹配、多匹配、异常等边界。延伸阅读测试与调试 RxJS 应用本文测试基础设施的完整来源包含collectionAssert实现、do调试与长堆栈支持创建与订阅简单 Observable 序列查询 Observable 序列按类别浏览操作符操作符 API 参考赞分享后端【免费下载链接】RxJSThe Reactive Extensions for JavaScript项目地址https://gitcode.com/gh_mirrors/rxj/RxJS点击查看免费下载相关推荐RxJS自定义操作符测试TDD开发流程RxJS自定义操作符测试TDD开发流程 你是否在开发RxJS操作符时遇到过难以调试的边界情况是否希望每一次代码变更都能安全可靠本文将通过TDD测试驱动开前端Javacord实战项目从零构建功能完整的Discord管理机器人Javacord实战项目从零构建功能完整的Discord管理机器人 你是否想拥有一个属于自己的Discord管理机器人却苦于找不到简单易用的Java开发库RxJava 自定义 Observable 操作符完全指南lift 序列操作符与 compose 转换操作符的源码级实战RxJava 自定义 Observable 操作符完全指南lift 序列操作符与 compose 转换操作符的源码级实战 在 RxJava 中编写自定义 Ob后端异步编程上一篇AssetRipper 快速上手指南从 Unity 游戏里完整提取资源下一篇革命性电话号码处理工具libphonenumber彻底解决全球号码格式混乱难题创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考
