Observable-RX 文档

简介

Observable-RX是一个基于观察者模式的JavaScript库，用于处理异步数据流。它提供了一种优雅的方式来处理异步数据流，使得代码更易于理解和维护。

有别于RxJS的使用，其中一点是，当发出了error事件，不需要retry（吐槽下retry也没用），next跟error可以同时不限制地使用，这也是自研的原因之一。

安装

使用npm安装：

1 npm install observable-rx

使用

创建Observable

要创建一个Observable，只需调用Observable构造函数即可。

1import { Observable } from 'observable-rx';
2
3const observable = new Observable();

订阅Observable

要订阅Observable，调用subscribe方法并传入一个观察者对象。观察者对象应该具有next、error方法，用于处理Observable发出的值和错误信号。

1import { Observable } from 'observable-rx';
2
3const observable = new Observable();
4
5// 订阅可观察对象
6const subscription = observable.subscribe({
7  next: (value) => console.log(value),
8  error: (value) => console.error(value),
9});
10
11// 取消订阅
12subscription.unsubscribe();

发布Observable

使用Observable的next方法，需要先创建一个Observable对象并调用subscribe方法进行订阅。

在观察者对象中，可以定义next方法来处理Observable发出的值。例如：

1import { Observable } from 'observable-rx';
2
3const observable = new Observable();
4
5// 订阅可观察对象
6observable.subscribe({
7  next: (value) => console.log(value),
8  error: (value) => console.error(value),
9});
10
11// 在需要的地方，调用Observable的next方法
12setTimeout(() => {
13    observable.next('hello world');
14}, 3000);

在上面的代码中，我们调用了Observable的next方法并传入了一个字符串作为参数。

当Observable发出这个值时，观察者对象的next方法将被调用并传入该值作为参数。

需要注意的是，如果Observable还没有被订阅，调用next方法将不会有任何效果。

如果你需要获取到上一次的值，在创建observable的时候，加上配置项relay：

1import { Observable } from 'observable-rx';
2
3// 传入options，配置relay为1
4const observable = new Observable({ options: { relay: 1 } });
5
6// 调用Observable的next方法
7observable.next('hello world');
8
9// 订阅可观察对象
10observable.subscribe({
11  next: (value) => console.log(value), // 会立即被调用一次
12  error: (value) => console.error(value),
13});

由于relay大于0，即使订阅在 observable.next('hello world') 之后执行，subscribe中的next回调也会被执行。

如果开启共享模式，即使订阅在值被发出之后执行，所有观察者对象的next方法都将被立即调用。

relay选项指定了Observable发出值的数量，而不是订阅的数量。

目前没有真正实现relay的次数，仅判断是否大于0，因此当relay大于0时，所有next回调都只会被立即执行1次。

抛出错误Observable

使用Observable的error方法，在观察者对象中，可以定义error方法来处理Observable发出的错误。例如：

1import { Observable } from 'observable-rx';
2
3const observable = new Observable({ options: { relay: 1 } });
4
5// 订阅可观察对象
6observable.subscribe({
7  next: (value) => console.log(value),
8  error: (value) => console.error(value),
9});
10
11// 出现异常时，调用Observable的error方法
12setTimeout(() => {
13    observable.error('something wrong');
14}, 5000);

在上面的代码中，我们调用了Observable的error方法并传入了一个字符串作为错误信息。当Observable发出这个值时，观察者对象的error方法将被调用并传入该值作为参数。

注意，错误不会因为relay配置而缓存下来，订阅时机在 observable.error('timeout') 执行之前不会获取历史错误信息。

共享Observable

Observable提供了一个share操作符，用于共享Observable的订阅，真正实现发布订阅模式，而不是一对一响应。

1import { Observable, share } from 'observable-rx';
2
3const observable = new Observable({ options: { relay: 1 } });
4
5// 该操作会重载部分observable的方法
6share()(observable);
7
8// 调用Observable的next方法
9observable.next('hello world');
10
11// 调用Observable的error方法
12setTimeout(() => {
13    observable.error('timeout');
14}, 5000);
15
16// 只能监听到next事件
17observable.subscribe((value) => console.log('subscribe-1: ', value));
18
19// 监听多种事件
20observable.subscribe({
21  next: (value) => console.log('subscribe-2-next: ', value),
22  error: (value) => console.error('subscribe-2-error: ', value),
23});
24
25// 再次调用Observable的next方法
26observable.next('hello again');

未开启共享模式时，重复调用 observable.subscribe 只会覆盖之前的回调。

API文档

构造函数

Observable

• new Observable(props?: Props): Observable

  创建一个新的Observable。

  参数：

  • props: 可选参数对象，包含以下属性：
    • initialData: 初始数据
    • options: 选项对象，包含以下属性：
      • relay: 是否在订阅时立即发布上一次的数据，传入大于0的relay值即可

  返回值：

  • Observable实例

  1import { Observable } from 'observable-rx';
  2
  3const observable = new Observable({ initialData: {}, options: { relay: 1 } });

实例方法

• subscribe(observer: Observer | Callback): Subscription

  订阅Observable。

  参数：

  • observer: 观察者对象，包含以下属性：
    • next: 数据处理函数，接收一个数据参数
    • error: 错误处理函数，接收一个错误参数

  或者：

  • callback: next回调，数据处理函数，接收一个数据参数

  返回值：

  • Subscription

  1const subscription = observable.subscribe({
  2    next: (value) => console.log(value),
  3    error: (value) => console.error(value),
  4});
  5
  6subscription.unsubscribe();

  或者

  1const subscription = observable.subscribe((value) => console.log(value));
  2
  3subscription.unsubscribe();

• next(data: any): void

  向 Observer 发送数据。

  参数：

  • data: 要发送的数据

  返回值：

  • void

  1observable.next('hello world');

• error(data: any): void

  向 Observer 发送错误信息。

  参数：

  • data: 要发送的错误

  返回值：

  • void

  1observable.error('something went wrong');

Operators

• share

  共享Observable的订阅。

  1import { share } from 'observable-rx';
  2
  3share()(observable);
  4
  5observable.subscribe((value) => console.log('subscribe shared observable 1: ', value));
  6
  7observable.subscribe({
  8    next: (value) => console.log('subscribe shared observer 2-next: ', value),
  9    error: (value) => console.error('subscribe shared observer 2-error: ', value),
  10});

特别说明

如您有意使用，欢迎联系本人yexinxie@163.com，这个小项目会根据您的意见修改并添加测试用例。

If you are interested in using it, please contact me at yexinxie@163.com. This small project will be modified and test cases will be added according to your feedback.

本库参考

• RxJS官方文档