响应式编程在前端领域的应用

import { ajax } from "rxjs/ajax";
import { map, retry, catchError } from "rxjs/operators";

const apiData = ajax("/api/data").pipe(
  // 可以在 catchError 之前使用 retry 操作符。它会订阅到原始的来源可观察对象，此处为重新发起 HTTP 请求
  retry(3), // 失败前会重试最多 3 次
  map((res) => {
    if (!res.response) {
      throw new Error("Value expected!");
    }
    return res.response;
  }),
  catchError((err) => of([]))
);

apiData.subscribe({
  next(x) {
    console.log("data: ", x);
  },
  error(err) {
    console.log("errors already caught... will not run");
  },
});

const observable = Rx.Observable.fromEvent(input, "input") // 监听 input 元素的 input 事件
  .map((e) => e.target.value) // 一旦发生，把事件对象 e 映射成 input 元素的值
  .filter((value) => value.length >= 1) // 接着过滤掉值长度小于 1 的
  .distinctUntilChanged() // 如果该值和过去最新的值相等，则忽略
  .subscribe(
    // subscribe 拿到数据
    (x) => console.log(x),
    (err) => console.error(err)
  );
// 订阅
observable.subscribe((x) => console.log(x));

let liveStreaming$ = Rx.Observable.interval(1000).take(5);

liveStreaming$.subscribe(
	data => console.log('subscriber from first second')
	err => console.log(err),
	() => console.log('completed')
)

setTimeout(() => {
	liveStreaming$.subscribe(
		data => console.log('subscriber from 2nd second')
		err => console.log(err),
		() => console.log('completed')
	)
}, 2000)
// 事实上两个订阅者接收到的值都是 0,1,2,3,4，此处为冷观察

let publisher$ = Rx.Observable.interval(1000).take(5).publish();

publisher$.subscribe(
	data => console.log('subscriber from first minute',data),
	err => console.log(err),
	() => console.log('completed')
)

setTimeout(() => {
    publisher$.subscribe(
        data => console.log('subscriber from 2nd minute', data),
        err => console.log(err),
        () => console.log('completed')
    )
}, 3000)

publisher$.connect();
// 第一个订阅者输出的是0,1,2,3,4，而第二个输出的是3,4，此处为热观察

# 1. merge
--1----2-----3--------4---
----a-----b----c---d------
           merge
--1-a--2--b--3-c---d--4---

# 2. combine
--1----2-----3--------4---
----a-----b-----c--d------
         combine
----1a-2a-2b-3b-3c-3d-4d--

const streamA1 = Rx.Observable.fromEvent(inputA1, "input"); // 监听 A1 单元格的 input 事件
const streamB2 = Rx.Observable.fromEvent(inputB2, "input"); // 监听 B2 单元格的 input 事件

const subscribe = combineLatest(streamA1, streamB2).subscribe((valueA1, valueB2) => {
	// 从 streamA1 和 streamB2 中获取最新发出的值
    return valueA1 + valaueB2;
});
// 获取函数计算结果
observable.subscribe((x) => console.log(x));

// timerOne 在 0 秒时发出第一个值，然后每 1 秒发送一次
const timerOne = timer(0, 1000).subscribe(x => {
	// 触发界面更新
});

var array = [1,2,3,4,5];

// 打印出每个项目
const subscription = Rx.Observable.from(array).subscribe(
	x => console.log('onNext: %s', x),
    e => console.log('onError: %s', e),
	() => console.log('onCompleted')
);

// => onNext: 1
// => onNext: 2
// => onNext: 3
// => onNext: 4
// => onNext: 5
// => onCompleted