简化的observable, 帮助理解observable.

obuservable.ts

type Unsubscribe = () => void

const NOOP: Unsubscribe = () => {}

interface Observer<T> {
    (t: T): void
}

interface Observable<T> {
    (ob: Observer<T>): Unsubscribe
}

class Subject<T> {
    private subscriptions: Set<Observer<T>>
    next: Observer<T> = (val: T) => {
        this.subscriptions.forEach(x => x(val))
    }
    // is complete really necessary? why not just let gc collect it?
    // complete() {
    //     this.subscriptions.clear()
    // }
    subscribe: Observable<T> = (observer: Observer<T>) => {
        this.subscriptions.add(observer)
        return () => {
            this.subscriptions.delete(observer)
        }
    }
}

type Operator<T1, T2> = (ob1: Observable<T1>) => Observable<T2>

type SecondOrderOperator<T1, T2, T3, T4> = (op1: Operator<T1, T2>) => Operator<T3, T4>

const catchError: <T1, T2>(handler: (error: unknown, unsubscribe: Unsubscribe) => Observable<T1>) => SecondOrderOperator<T1, T2, T1, T2> = handler => op1 => {
    return ob1 => {
        const ob2 = op1(obv1 => {
            const unsub = ob1(val => {
                try {
                    obv1(val)
                } catch (err) {
                    handler(err, unsub)
                }
            })
            return unsub
        })
        return ob2
    }
}

const fromInterval: (ms: number) => Observable<number> = ms => observer => {
    const unsub = () => {
        clearInterval(i)
        console.log("interval cleared")
    }
    // const next = observer(() => {
    //     clearInterval(i)
    // })
    let counter = 0
    const i = setInterval(() => {
        observer(counter++)
    }, ms)
    return unsub
}

const take: <T1>(num: number) => Operator<T1, T1> = num => source => {
    var cur = 0
    return sink => {
        const unsubscribe = source(val => {
            cur++
            if (cur <= num) {
                sink(val)
            } else {
                unsubscribe()
            }
        })
        return unsubscribe
    }
}

const of: <T>(val: T) => Observable<T> = val => next => {
    next(val)
    return NOOP
}

const map: <T1, T2>(fmap: (t: T1) => T2) => Operator<T1, T2> = fmap => source => sink => source(val => sink(fmap(val)))

const pipe: <T1, T2, T3>(op1: Operator<T1, T2>, op2: Operator<T2, T3>) => Operator<T1, T3> = (op1, op2) => ob1 => op2(op1(ob1))

const unsub = pipe(
    () => fromInterval(1000),
    map(num => {
        if (num >= 5) {
            console.log("we have unsubed")
            unsub()
        }
        return num + 1
    })
)(of(null))(val => {
    console.log(val)
})

console.log("are we here")

哈哈哈哈这居然他妈能跑