【draft】RxJS学习记录

Copycopy code to clipboard
1const sourceOb = from([1, 2, 3]);
2
3const sub = new Subject();
4
5const multicasted = sourceOb.pipe(multicast(sub));
6
7// 实际上即是sub.subscribe
8multicasted.subscribe((x) => console.log(`1-${x}`));
9multicasted.subscribe((x) => console.log(`2-${x}`));
10
11sourceOb.subscribe(sub);

Copycopy code to clipboard
1const source = interval(500);
2const subject = new Subject();
3const refCounted = source.pipe(multicast(subject), refCount());
4
5let subscription1: Subscription, subscription2: Subscription;
6
7console.log("observerA subscribed");
8// 会自动开始共享ob 因为第一个订阅者出现了
9subscription1 = refCounted.subscribe({
10  next: (v) => console.log(`observerA: ${v}`),
11});
12
13setTimeout(() => {
14  console.log("observerB subscribed");
15  subscription2 = refCounted.subscribe({
16    next: (v) => console.log(`observerB: ${v}`),
17  });
18}, 600);
19
20setTimeout(() => {
21  console.log("observerA unsubscribed");
22  subscription1.unsubscribe();
23}, 1200);
24
25// 取消共享 因为没有观察者了
26setTimeout(() => {
27  console.log("observerB unsubscribed");
28  subscription2.unsubscribe();
29}, 2000);

Copycopy code to clipboard
1const subject = new BehaviorSubject(0); // 0 is the initial value
2
3subject.subscribe({
4  next: (v) => console.log(`observerA: ${v}`),
5});
6
7subject.next(1);
8subject.next(2);
9
10subject.subscribe({
11  next: (v) => console.log(`observerB: ${v}`),
12});
13
14subject.next(3);
15
16observerA: 0;
17observerA: 1;
18observerA: 2;
19observerB: 2;
20observerA: 3;
21observerB: 3;

Copycopy code to clipboard
1subject.subscribe({
2  next: (v) => console.log(`observerA: ${v}`),
3});
4
5subject.next(1);
6subject.next(2);
7subject.next(3);
8subject.next(4);
9
10subject.subscribe({
11  next: (v) => console.log(`observerB: ${v}`),
12});
13
14subject.next(5);
15
16// observerA: 1
17// observerA: 2
18// observerA: 3
19// observerA: 4
20// observerB: 2
21// observerB: 3
22// observerB: 4
23// observerA: 5
24// observerB: 5

Copycopy code to clipboard
1const subject = new AsyncSubject();
2
3subject.subscribe({
4  next: (v) => console.log(`observerA: ${v}`),
5});
6
7subject.next(1);
8subject.next(2);
9subject.next(3);
10subject.next(4);
11
12subject.subscribe({
13  next: (v) => console.log(`observerB: ${v}`),
14});
15
16subject.next(5);
17subject.complete();
18
19// Logs:
20// observerA: 5
21// observerB: 5

Copycopy code to clipboard
1import { Observable } from "rxjs";
2import { createTodoItem } from "./lib";
3
4const $input = <HTMLInputElement>document.querySelector(".todo-val");
5const $list = <HTMLUListElement>document.querySelector(".list-group");
6const $add = document.querySelector(".button-add");
7
8const enter$ = Observable.fromEvent<KeyboardEvent>($input, "keydown").filter(
9  // 监听enter键作为一个单独的流
10  (r) => r.keyCode === 13
11);
12
13// add键的点击同理
14const clickAdd$ = Observable.fromEvent<MouseEvent>($add, "click");
15
16// 将enter与add点击监听合并为一个流
17const input$ = enter$.merge(clickAdd$);
18
19const item$ = input$
20  // 提取输入值
21  .map(() => $input.value)
22  // 去空
23  .filter((r) => r !== "")
24  // 返回一个DOM片段
25  .map(createTodoItem)
26  // 使用do来执行副作用/DOM操作等
27  .do((ele: HTMLLIElement) => {
28    $list.appendChild(ele);
29    // 清空输入框
30    $input.value = "";
31  })
32  .publishReplay(1)
33  .refCount();
34
35const toggle$ = item$
36  // mergeMap 对每个值进行一次映射 并使用mergeAll展平最后的结果
37  .mergeMap(($todoItem) =>
38    // 将item映射到item的点击事件
39    Observable.fromEvent<MouseEvent>($todoItem, "click")
40      // 仅关心item被点击
41      .filter((e) => e.target === $todoItem)
42      // 再将每次点击映射回item
43      .mapTo($todoItem)
44  )
45  .do(($todoItem: HTMLElement) => {
46    // 处理样式(标识状态)
47    if ($todoItem.classList.contains("done")) {
48      $todoItem.classList.remove("done");
49    } else {
50      $todoItem.classList.add("done");
51    }
52  });
53
54const remove$ = item$
55  .mergeMap(($todoItem) => {
56    const $removeButton = $todoItem.querySelector(".button-remove");
57    return Observable.fromEvent($removeButton, "click").mapTo($todoItem);
58  })
59  .do(($todoItem: HTMLElement) => {
60    // 从 DOM 上移掉 todo item
61    const $parent = $todoItem.parentNode;
62    $parent.removeChild($todoItem);
63  });
64
65const app$ = toggle$.merge(remove$).do((r) => console.log(r));
66
67app$.subscribe();
68
69// lib
70export const createTodoItem = (val: string) => {
71  const result = <HTMLLIElement>document.createElement("LI");
72  result.classList.add("list-group-item");
73  const innerHTML = `
74    ${val}
75    <button type="button" class="btn btn-default button-remove" aria-label="right Align">
76      <span class="glyphicon glyphicon-remove" aria-hidden="true"></span>
77    </button>
78  `;
79  result.innerHTML = innerHTML;
80  return result;
81};

Copycopy code to clipboard
1import { Observable, Subject } from "rxjs";
2import {
3  createTodoItem,
4  mockToggle,
5  mockHttpPost,
6  search,
7  HttpResponse,
8} from "./lib";
9
10const $input = <HTMLInputElement>document.querySelector(".todo-val");
11const $list = <HTMLUListElement>document.querySelector(".list-group");
12const $add = document.querySelector(".button-add");
13
14// 后面的 search$ 与 enter 应该时从同一个 Observable 中转换出来，这里将 input 事件的 Observable publish 成 muticast
15const type$ = Observable.fromEvent<KeyboardEvent>($input, "keydown")
16  .publish()
17  .refCount();
18
19const search$ = type$
20  .debounceTime(200)
21  .filter((evt) => evt.keyCode !== 13)
22  .map((result) => (<HTMLInputElement>result.target).value)
23  .switchMap(search)
24  .do((result: HttpResponse | null) => {
25    const actived = document.querySelectorAll(".active");
26    Array.prototype.forEach.call(actived, (item: HTMLElement) => {
27      item.classList.remove("active");
28    });
29    if (result) {
30      const item = document.querySelector(`.todo-item-${result._id}`);
31      item.classList.add("active");
32    }
33  });
34
35const enter$ = type$.filter((r) => r.keyCode === 13);
36
37const clickAdd$ = Observable.fromEvent<MouseEvent>($add, "click");
38
39const input$ = enter$.merge(clickAdd$);
40
41const clearInputSubject$ = new Subject<any>();
42
43const item$ = input$
44  .map(() => $input.value)
45  .filter((r) => r !== "")
46  // 使输入框值不变时无法走到请求取消的环节
47  // 返回与之前源ob都不同的值
48  // .distinct()
49  // 避免在输入值相同时也过滤掉ob
50  // flushes会清空缓存：flushes会被订阅，并在完成时清空缓存
51  .distinct(null, clearInputSubject$)
52  // 当ob内部流动的值同样是ob时  订阅最新的一个 将其的值传给下一个操作符 并取消对上一个的订阅
53  // 实际上就是先switch然后map
54  .switchMap(mockHttpPost)
55  .map(createTodoItem)
56  .do((ele: HTMLLIElement) => {
57    $list.appendChild(ele);
58    $input.value = "";
59  })
60  .publishReplay(1)
61  .refCount();
62
63const toggle$ = item$
64  .mergeMap(($todoItem) => {
65    return (
66      Observable.fromEvent<MouseEvent>($todoItem, "click")
67        // 300ms内只会发出一次
68        .debounceTime(300)
69        .filter((e) => e.target === $todoItem)
70        .mapTo({
71          data: {
72            _id: $todoItem.dataset["id"],
73            isDone: $todoItem.classList.contains("done"),
74          },
75          $todoItem,
76        })
77    );
78  })
79  .switchMap((result) =>
80    mockToggle(result.data._id, result.data.isDone)
81      // 映射回原对象
82      .mapTo(result.$todoItem)
83  );
84
85const remove$ = item$
86  .mergeMap(($todoItem) => {
87    const $removeButton = $todoItem.querySelector(".button-remove");
88    return Observable.fromEvent($removeButton, "click").mapTo($todoItem);
89  })
90  .do(($todoItem: HTMLElement) => {
91    // 从 DOM 上移掉 todo item
92    const $parent = $todoItem.parentNode;
93    $parent.removeChild($todoItem);
94  });
95
96const app$ = toggle$.merge(remove$, search$).do((r) => {
97  console.log(r);
98});
99
100app$.subscribe();

Copycopy code to clipboard
1import { Observable, Subscriber, Subject } from "rxjs";
2// spark-md5 没有第三方 .d.ts 文件，这里用 commonjs 风格的 require 它
3// 如果未再 tsconfig.json 中设置 noImplicitAny: true 且 TypeScript 版本大于 2.1 则也可以用
4// import * as SparkMD5 from 'spark-md5' 的方式引用
5const SparkMD5 = require("spark-md5");
6// @warn memory leak
7const $attachment = document.querySelector(".attachment");
8const $progressBar = document.querySelector(".progress-bar") as HTMLElement;
9const apiHost = "http://127.0.0.1:5000/api";
10
11interface FileInfo {
12  fileSize: number;
13  fileMD5: string;
14  lastUpdated: string;
15  fileName: string;
16}
17
18interface ChunkMeta {
19  fileSize: number;
20  chunkSize: number;
21  chunks: number;
22  fileKey: string;
23}
24
25type Action = "pause" | "resume" | "progress" | "complete";
26
27export class FileUploader {
28  // 有文件进入时触发
29  // 原始流
30  private file$ = Observable.fromEvent($attachment, "change")
31    .map((r: Event) => (r.target as HTMLInputElement).files[0])
32    .filter((f) => !!f);
33
34  private click$ = Observable.fromEvent($attachment, "click")
35    // 过滤子节点冒泡
36    .map((e: Event) => e.target)
37    .filter((e: HTMLElement) => e === $attachment)
38    // 1-2-3 上传-暂停-继续
39    // scan会不断生成这几个状态
40    .scan((acc: number, val: HTMLElement) => {
41      if (val.classList.contains("glyphicon-paperclip")) {
42        return 1;
43      }
44      if (acc === 2) {
45        return 3;
46      }
47      return 2;
48    }, 3)
49    .filter((v) => v !== 1)
50    // 根据状态改变icon样式
51    .do((v) => {
52      console.log(v);
53      if (v === 2) {
54        this.action$.next({ name: "pause" });
55        $attachment.classList.remove("glyphicon-pause");
56        $attachment.classList.add("glyphicon-play");
57      } else {
58        this.action$.next({ name: "resume" });
59        this.buildPauseIcon();
60      }
61    })
62    .map((v) => ({ action: v === 2 ? "PAUSE" : "RESUME", payload: null }));
63
64  private action$ = new Subject<{
65    name: Action;
66    payload?: any;
67  }>();
68
69  private pause$ = this.action$.filter((ac) => ac.name === "pause");
70  private resume$ = this.action$.filter((ac) => ac.name === "resume");
71
72  // 进度流
73  private progress$ = this.action$
74    .filter((action) => action.name === "progress")
75    .map((action) => action.payload)
76    .distinctUntilChanged((x: number, y: number) => x - y >= 0)
77    .do((r) => {
78      const percent = Math.round(r * 100);
79      $progressBar.style.width = `${percent}%`;
80      $progressBar.firstElementChild.textContent = `${
81        percent > 1 ? percent - 1 : percent
82      } %`;
83    })
84    .map((r) => ({ action: "PROGRESS", payload: r }));
85
86  public uploadStream$ = this.file$
87    .switchMap(this.readFileInfo)
88    .switchMap((i) =>
89      Observable.ajax
90        // 获取分片信息(并不是实际上传)
91        .post(`${apiHost}/upload/chunk`, i.fileinfo)
92        .map((r) => {
93          // 按照返回的分片信息进行分片
94          const blobs = this.slice(
95            i.file,
96            r.response.chunks,
97            r.response.chunkSize
98          );
99          return { blobs, chunkMeta: r.response };
100        })
101    )
102    // 创建暂停按钮
103    .do(() => this.buildPauseIcon())
104    .switchMap(({ blobs, chunkMeta }) => {
105      const uploaded: number[] = [];
106      const dists = blobs.map((blob, index) => {
107        let currentLoaded = 0;
108        return this.uploadChunk(chunkMeta, index, blob).do((r) => {
109          currentLoaded = r.loaded / chunkMeta.fileSize;
110          uploaded[index] = currentLoaded;
111          const percent = uploaded.reduce((acc, val) => acc + (val ? val : 0));
112          // 计算进度
113          this.action$.next({ name: "progress", payload: percent });
114        });
115      });
116
117      // 并发上传所有分片(并发度3)
118      const uploadStream = Observable.from(dists).mergeAll(this.concurrency);
119
120      // 所有分片上传完毕后输出值
121      // 再映射到本次切片的元数据chunkMeta
122      return Observable.forkJoin(uploadStream).mapTo(chunkMeta);
123    })
124    // 上传分片
125    .switchMap((r: ChunkMeta) =>
126      Observable.ajax
127        .post(`${apiHost}/upload/chunk/${r.fileKey}`)
128        // 将请求映射到上传状态
129        .mapTo({
130          action: "UPLOAD_SUCCESS",
131          payload: r,
132        })
133    )
134    .do(() => {
135      $progressBar.firstElementChild.textContent = "100 %";
136      // restore icon
137      $attachment.classList.remove("glyphicon-pause");
138      $attachment.classList.add("glyphicon-paperclip");
139      ($attachment.firstElementChild as HTMLInputElement).disabled = false;
140    })
141    // 与过程流 点击流合并
142    .merge(this.progress$, this.click$);
143
144  constructor(private concurrency = 3) {}
145
146  // side effect
147  private buildPauseIcon() {
148    $attachment.classList.remove("glyphicon-paperclip");
149    $attachment.classList.add("glyphicon-pause");
150    ($attachment.firstElementChild as HTMLInputElement).disabled = true;
151  }
152
153  // 读取文件信息
154  // 拿到文件流后会附带文件信息与MD5信息
155  // 用于uploadSream$的第一次switchMap
156  private readFileInfo(
157    file: File
158  ): Observable<{ file: File; fileinfo: FileInfo }> {
159    const reader = new FileReader();
160    const spark = new SparkMD5.ArrayBuffer();
161    reader.readAsArrayBuffer(file);
162    return Observable.create(
163      (observer: Subscriber<{ file: File; fileinfo: FileInfo }>) => {
164        reader.onload = (e: Event) => {
165          spark.append((e.target as FileReader).result);
166          const fileMD5 = spark.end();
167          observer.next({
168            file,
169            fileinfo: {
170              fileMD5,
171              fileSize: file.size,
172              lastUpdated: file.lastModifiedDate.toISOString(),
173              fileName: file.name,
174            },
175          });
176          observer.complete();
177        };
178        return () => {
179          if (!reader.result) {
180            console.warn("read file aborted");
181            reader.abort();
182          }
183        };
184      }
185    );
186  }
187
188  private slice(file: File, n: number, chunkSize: number): Blob[] {
189    const result: Blob[] = [];
190    for (let i = 0; i < n; i++) {
191      const startSize = i * chunkSize;
192      const slice = file.slice(
193        startSize,
194        i === n - 1 ? startSize + (file.size - startSize) : (i + 1) * chunkSize
195      );
196      result.push(slice);
197    }
198    return result;
199  }
200
201  private uploadChunk(
202    meta: ChunkMeta,
203    index: number,
204    blob: Blob
205  ): Observable<ProgressEvent> {
206    const host = `${apiHost}/upload/chunk/${meta.fileKey}?chunk=${
207      index + 1
208    }&chunks=${meta.chunks}`;
209    return Observable.create((subscriber: Subscriber<ProgressEvent>) => {
210      const ajax$ = Observable.ajax({
211        url: host,
212        body: blob,
213        method: "post",
214        crossDomain: true,
215        headers: { "Content-Type": "application/octet-stream" },
216        // 进度获取
217        progressSubscriber: subscriber,
218      })
219        // 在暂停流有输出时 停止输出值
220        .takeUntil(this.pause$)
221        // 在恢复流有输出时 重复
222        .repeatWhen(() => this.resume$);
223      const subscription = ajax$.subscribe();
224      return () => subscription.unsubscribe();
225    }).retryWhen(() => this.resume$);
226  }
227}