RxJS学习记录

const sourceOb = from([1, 2, 3]);
const sub = new Subject();
const multicasted = sourceOb.pipe(multicast(sub));
// 实际上即是sub.subscribe
multicasted.subscribe((x) => console.log(`1-${x}`));
multicasted.subscribe((x) => console.log(`2-${x}`));
sourceOb.subscribe(sub);

const source = interval(500);
const subject = new Subject();
const refCounted = source.pipe(multicast(subject), refCount());
let subscription1: Subscription, subscription2: Subscription;
console.log('observerA subscribed');
// 会自动开始共享ob 因为第一个订阅者出现了
subscription1 = refCounted.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
});
setTimeout(() => {
  console.log('observerB subscribed');
  subscription2 = refCounted.subscribe({
    next: (v) => console.log(`observerB: ${v}`),
  });
}, 600);
setTimeout(() => {
  console.log('observerA unsubscribed');
  subscription1.unsubscribe();
}, 1200);
// 取消共享 因为没有观察者了
setTimeout(() => {
  console.log('observerB unsubscribed');
  subscription2.unsubscribe();
}, 2000);

const subject = new BehaviorSubject(0); // 0 is the initial value
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`),
});
subject.next(1);
subject.next(2);
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`),
});
subject.next(3);
observerA: 0
observerA: 1
observerA: 2
observerB: 2
observerA: 3
observerB: 3

subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
subject.next(5);
// observerA: 1
// observerA: 2
// observerA: 3
// observerA: 4
// observerB: 2
// observerB: 3
// observerB: 4
// observerA: 5
// observerB: 5

const subject = new AsyncSubject();
subject.subscribe({
  next: (v) => console.log(`observerA: ${v}`)
});
subject.next(1);
subject.next(2);
subject.next(3);
subject.next(4);
subject.subscribe({
  next: (v) => console.log(`observerB: ${v}`)
});
subject.next(5);
subject.complete();
// Logs:
// observerA: 5
// observerB: 5

import { Observable } from "rxjs";
import { createTodoItem } from "./lib";
const $input = <HTMLInputElement>document.querySelector(".todo-val");
const $list = <HTMLUListElement>document.querySelector(".list-group");
const $add = document.querySelector(".button-add");
const enter$ = Observable.fromEvent<KeyboardEvent>($input, "keydown").filter(
  // 监听enter键作为一个单独的流
  (r) => r.keyCode === 13
);
// add键的点击同理
const clickAdd$ = Observable.fromEvent<MouseEvent>($add, "click");
// 将enter与add点击监听合并为一个流
const input$ = enter$.merge(clickAdd$);
const item$ = input$
  // 提取输入值
  .map(() => $input.value)
  // 去空
  .filter((r) => r !== "")
  // 返回一个DOM片段
  .map(createTodoItem)
  // 使用do来执行副作用/DOM操作等
  .do((ele: HTMLLIElement) => {
    $list.appendChild(ele);
    // 清空输入框
    $input.value = "";
  })
  .publishReplay(1)
  .refCount();
const toggle$ = item$
  // mergeMap 对每个值进行一次映射 并使用mergeAll展平最后的结果
  .mergeMap(($todoItem) =>
    // 将item映射到item的点击事件
    Observable.fromEvent<MouseEvent>($todoItem, "click")
      // 仅关心item被点击
      .filter((e) => e.target === $todoItem)
      // 再将每次点击映射回item
      .mapTo($todoItem)
  )
  .do(($todoItem: HTMLElement) => {
    // 处理样式(标识状态)
    if ($todoItem.classList.contains("done")) {
      $todoItem.classList.remove("done");
    } else {
      $todoItem.classList.add("done");
    }
  });
const remove$ = item$
  .mergeMap(($todoItem) => {
    const $removeButton = $todoItem.querySelector(".button-remove");
    return Observable.fromEvent($removeButton, "click").mapTo($todoItem);
  })
  .do(($todoItem: HTMLElement) => {
    // 从 DOM 上移掉 todo item
    const $parent = $todoItem.parentNode;
    $parent.removeChild($todoItem);
  });
const app$ = toggle$.merge(remove$).do((r) => console.log(r));
app$.subscribe();
// lib
export const createTodoItem = (val: string) => {
  const result = <HTMLLIElement>document.createElement("LI");
  result.classList.add("list-group-item");
  const innerHTML = `
    ${val}
    <button type="button" class="btn btn-default button-remove" aria-label="right Align">
      <span class="glyphicon glyphicon-remove" aria-hidden="true"></span>
    </button>
  `;
  result.innerHTML = innerHTML;
  return result;
};

import { Observable, Subject } from "rxjs";
import {
  createTodoItem,
  mockToggle,
  mockHttpPost,
  search,
  HttpResponse,
} from "./lib";
const $input = <HTMLInputElement>document.querySelector(".todo-val");
const $list = <HTMLUListElement>document.querySelector(".list-group");
const $add = document.querySelector(".button-add");
// 后面的 search$ 与 enter 应该时从同一个 Observable 中转换出来，这里将 input 事件的 Observable publish 成 muticast
const type$ = Observable.fromEvent<KeyboardEvent>($input, "keydown")
  .publish()
  .refCount();
const search$ = type$
  .debounceTime(200)
  .filter((evt) => evt.keyCode !== 13)
  .map((result) => (<HTMLInputElement>result.target).value)
  .switchMap(search)
  .do((result: HttpResponse | null) => {
    const actived = document.querySelectorAll(".active");
    Array.prototype.forEach.call(actived, (item: HTMLElement) => {
      item.classList.remove("active");
    });
    if (result) {
      const item = document.querySelector(`.todo-item-${result._id}`);
      item.classList.add("active");
    }
  });
const enter$ = type$.filter((r) => r.keyCode === 13);
const clickAdd$ = Observable.fromEvent<MouseEvent>($add, "click");
const input$ = enter$.merge(clickAdd$);
const clearInputSubject$ = new Subject<any>();
const item$ = input$
  .map(() => $input.value)
  .filter((r) => r !== "")
  // 使输入框值不变时无法走到请求取消的环节
  // 返回与之前源ob都不同的值
  // .distinct()
  // 避免在输入值相同时也过滤掉ob
  // flushes会清空缓存：flushes会被订阅，并在完成时清空缓存
  .distinct(null, clearInputSubject$)
  // 当ob内部流动的值同样是ob时  订阅最新的一个 将其的值传给下一个操作符 并取消对上一个的订阅
  // 实际上就是先switch然后map
  .switchMap(mockHttpPost)
  .map(createTodoItem)
  .do((ele: HTMLLIElement) => {
    $list.appendChild(ele);
    $input.value = "";
  })
  .publishReplay(1)
  .refCount();
const toggle$ = item$
  .mergeMap(($todoItem) => {
    return (
      Observable.fromEvent<MouseEvent>($todoItem, "click")
        // 300ms内只会发出一次
        .debounceTime(300)
        .filter((e) => e.target === $todoItem)
        .mapTo({
          data: {
            _id: $todoItem.dataset["id"],
            isDone: $todoItem.classList.contains("done"),
          },
          $todoItem,
        })
    );
  })
  .switchMap((result) =>
    mockToggle(result.data._id, result.data.isDone)
      // 映射回原对象
      .mapTo(result.$todoItem)
  );
const remove$ = item$
  .mergeMap(($todoItem) => {
    const $removeButton = $todoItem.querySelector(".button-remove");
    return Observable.fromEvent($removeButton, "click").mapTo($todoItem);
  })
  .do(($todoItem: HTMLElement) => {
    // 从 DOM 上移掉 todo item
    const $parent = $todoItem.parentNode;
    $parent.removeChild($todoItem);
  });
const app$ = toggle$.merge(remove$, search$).do((r) => {
  console.log(r);
});
app$.subscribe();

import { Observable, Subscriber, Subject } from "rxjs";
// spark-md5 没有第三方 .d.ts 文件，这里用 commonjs 风格的 require 它
// 如果未再 tsconfig.json 中设置 noImplicitAny: true 且 TypeScript 版本大于 2.1 则也可以用
// import * as SparkMD5 from 'spark-md5' 的方式引用
const SparkMD5 = require("spark-md5");
// @warn memory leak
const $attachment = document.querySelector(".attachment");
const $progressBar = document.querySelector(".progress-bar") as HTMLElement;
const apiHost = "http://127.0.0.1:5000/api";
interface FileInfo {
  fileSize: number;
  fileMD5: string;
  lastUpdated: string;
  fileName: string;
}
interface ChunkMeta {
  fileSize: number;
  chunkSize: number;
  chunks: number;
  fileKey: string;
}
type Action = "pause" | "resume" | "progress" | "complete";
export class FileUploader {
  // 有文件进入时触发
  // 原始流
  private file$ = Observable.fromEvent($attachment, "change")
    .map((r: Event) => (r.target as HTMLInputElement).files[0])
    .filter((f) => !!f);
  private click$ = Observable.fromEvent($attachment, "click")
    // 过滤子节点冒泡
    .map((e: Event) => e.target)
    .filter((e: HTMLElement) => e === $attachment)
    // 1-2-3 上传-暂停-继续
    // scan会不断生成这几个状态
    .scan((acc: number, val: HTMLElement) => {
      if (val.classList.contains("glyphicon-paperclip")) {
        return 1;
      }
      if (acc === 2) {
        return 3;
      }
      return 2;
    }, 3)
    .filter((v) => v !== 1)
    // 根据状态改变icon样式
    .do((v) => {
      console.log(v);
      if (v === 2) {
        this.action$.next({ name: "pause" });
        $attachment.classList.remove("glyphicon-pause");
        $attachment.classList.add("glyphicon-play");
      } else {
        this.action$.next({ name: "resume" });
        this.buildPauseIcon();
      }
    })
    .map((v) => ({ action: v === 2 ? "PAUSE" : "RESUME", payload: null }));
  private action$ = new Subject<{
    name: Action;
    payload?: any;
  }>();
  private pause$ = this.action$.filter((ac) => ac.name === "pause");
  private resume$ = this.action$.filter((ac) => ac.name === "resume");
  // 进度流
  private progress$ = this.action$
    .filter((action) => action.name === "progress")
    .map((action) => action.payload)
    .distinctUntilChanged((x: number, y: number) => x - y >= 0)
    .do((r) => {
      const percent = Math.round(r * 100);
      $progressBar.style.width = `${percent}%`;
      $progressBar.firstElementChild.textContent = `${
        percent > 1 ? percent - 1 : percent
      } %`;
    })
    .map((r) => ({ action: "PROGRESS", payload: r }));
  public uploadStream$ = this.file$
    .switchMap(this.readFileInfo)
    .switchMap((i) =>
      Observable.ajax
        // 获取分片信息(并不是实际上传)
        .post(`${apiHost}/upload/chunk`, i.fileinfo)
        .map((r) => {
          // 按照返回的分片信息进行分片
          const blobs = this.slice(
            i.file,
            r.response.chunks,
            r.response.chunkSize
          );
          return { blobs, chunkMeta: r.response };
        })
    )
    // 创建暂停按钮
    .do(() => this.buildPauseIcon())
    .switchMap(({ blobs, chunkMeta }) => {
      const uploaded: number[] = [];
      const dists = blobs.map((blob, index) => {
        let currentLoaded = 0;
        return this.uploadChunk(chunkMeta, index, blob).do((r) => {
          currentLoaded = r.loaded / chunkMeta.fileSize;
          uploaded[index] = currentLoaded;
          const percent = uploaded.reduce((acc, val) => acc + (val ? val : 0));
          // 计算进度
          this.action$.next({ name: "progress", payload: percent });
        });
      });
      // 并发上传所有分片(并发度3)
      const uploadStream = Observable.from(dists).mergeAll(this.concurrency);
      // 所有分片上传完毕后输出值
      // 再映射到本次切片的元数据chunkMeta
      return Observable.forkJoin(uploadStream).mapTo(chunkMeta);
    })
    // 上传分片
    .switchMap((r: ChunkMeta) =>
      Observable.ajax
        .post(`${apiHost}/upload/chunk/${r.fileKey}`)
        // 将请求映射到上传状态
        .mapTo({
          action: "UPLOAD_SUCCESS",
          payload: r,
        })
    )
    .do(() => {
      $progressBar.firstElementChild.textContent = "100 %";
      // restore icon
      $attachment.classList.remove("glyphicon-pause");
      $attachment.classList.add("glyphicon-paperclip");
      ($attachment.firstElementChild as HTMLInputElement).disabled = false;
    })
    // 与过程流 点击流合并
    .merge(this.progress$, this.click$);
  constructor(private concurrency = 3) {}
  // side effect
  private buildPauseIcon() {
    $attachment.classList.remove("glyphicon-paperclip");
    $attachment.classList.add("glyphicon-pause");
    ($attachment.firstElementChild as HTMLInputElement).disabled = true;
  }
  // 读取文件信息
  // 拿到文件流后会附带文件信息与MD5信息
  // 用于uploadSream$的第一次switchMap
  private readFileInfo(
    file: File
  ): Observable<{ file: File; fileinfo: FileInfo }> {
    const reader = new FileReader();
    const spark = new SparkMD5.ArrayBuffer();
    reader.readAsArrayBuffer(file);
    return Observable.create(
      (observer: Subscriber<{ file: File; fileinfo: FileInfo }>) => {
        reader.onload = (e: Event) => {
          spark.append((e.target as FileReader).result);
          const fileMD5 = spark.end();
          observer.next({
            file,
            fileinfo: {
              fileMD5,
              fileSize: file.size,
              lastUpdated: file.lastModifiedDate.toISOString(),
              fileName: file.name,
            },
          });
          observer.complete();
        };
        return () => {
          if (!reader.result) {
            console.warn("read file aborted");
            reader.abort();
          }
        };
      }
    );
  }
  private slice(file: File, n: number, chunkSize: number): Blob[] {
    const result: Blob[] = [];
    for (let i = 0; i < n; i++) {
      const startSize = i * chunkSize;
      const slice = file.slice(
        startSize,
        i === n - 1 ? startSize + (file.size - startSize) : (i + 1) * chunkSize
      );
      result.push(slice);
    }
    return result;
  }
  private uploadChunk(
    meta: ChunkMeta,
    index: number,
    blob: Blob
  ): Observable<ProgressEvent> {
    const host = `${apiHost}/upload/chunk/${meta.fileKey}?chunk=${
      index + 1
    }&chunks=${meta.chunks}`;
    return Observable.create((subscriber: Subscriber<ProgressEvent>) => {
      const ajax$ = Observable.ajax({
        url: host,
        body: blob,
        method: "post",
        crossDomain: true,
        headers: { "Content-Type": "application/octet-stream" },
        // 进度获取
        progressSubscriber: subscriber,
      })
        // 在暂停流有输出时 停止输出值
        .takeUntil(this.pause$)
        // 在恢复流有输出时 重复
        .repeatWhen(() => this.resume$);
      const subscription = ajax$.subscribe();
      return () => subscription.unsubscribe();
    }).retryWhen(() => this.resume$);
  }
}