簡單模拟實作Rxjs Observable

1.先定義類型

export type Observer = {
    next: (any) => void,
    complete?: (any) => void,
}

export interface OnSubscribeAction {
    (observer: Observer): void;
}
      

　export interface mapFun {

      (x: any): any

  }

2.自定義類

export class MyObservable {
    private subscribeAction: OnSubscribeAction;
    constructor(subscribeAction: OnSubscribeAction) {
        this.subscribeAction = subscribeAction;
    }

    subscribe(observer: Observer) {
        return this.subscribeAction(observer)
    }
}      

3.執行個體

const onSubscribe: OnSubscribeAction = (observer: Observer) => {
      observer.next(1);
      observer.next(2);
      observer.next(3);
}

const theObserver = {
   next: console.log,
}

const source$ =  new MyObservable(onSubscribe);
source$.subscribe(theObserver);      

結果一次輸出1,2,3

4.增加map操作符

export class MyObservable {
    private subscribeAction: OnSubscribeAction;
    constructor(subscribeAction: OnSubscribeAction) {
        this.subscribeAction = subscribeAction;
    }

    subscribe(observer: Observer) {
        return this.subscribeAction(observer)
    }
　　//rxjs 的每個操作符都會傳回一個新的Observable
    map(fun: mapFun) {
         const nextObserver =  new MyObservable((observer: Observer) => {
            this.subscribe({
                next(item){
                    observer.next(fun(item))
                }
            })
         })
         return nextObserver
    }
}      

5.

const onSubscribe: OnSubscribeAction = (observer: Observer) => {
      observer.next(1);
      observer.next(2);
      observer.next(3);
    }

    const theObserver = {
      next: console.log,
    }

    const source$ =  MyObservable.create(onSubscribe);
    source$.map(x => x * 2).map(x => x + 1).subscribe(theObserver);      

結果一次輸出3，5，7