rx.js

https://www.youtube.com/watch?v=uQ1zhJHclvs

function getMeData(ob) {
  [10,20,30].forEach(ob.next)
}
giveMeData => subscribe(ob)

const observable = {
 subscribe: (ob) {
    [10,20,30].forEach(ob.next)
  }
}
//  setter => setter
function map(transform) {
    const inputObservable = this
    
    const outputObservable = createObservable(function subscribe(observe) {
        inputObservable.subscribe({
            next(val) {
                const y = transform(val)
                observe.next(y)
            },
            err(error) {
                console.error(error)
            },
            complete() {
                console.log('done')
            }
        })
    })
    return outputObservable
}

function filter(condition) {
    const inputObservable = this
    
    const outputObservable = createObservable(function subscribe(observe) {
        inputObservable.subscribe({
            next(val) {
                if(condition(val)){
                    observe.next(val)
                }
            },
            err(error) {
                console.error(error)
            },
            complete() {
                console.log('done')
            }
        })
    })
    return outputObservable
}


var createObservable = function(subscribe) {
    return ({
       subscribe: subscribe,
       map: map,
       filter: filter
    })
}

var observe = {
    next(val) {
        console.log(val)
    },
    err(error) {
        console.error(error)
    },
    complete() {
        console.log('done')
    }
}

var arrayObservable = createObservable(function subscribe(ob) {
    [10,20,30].forEach(ob.next)
    ob.complete()
})

arrayObservable         // {subscribe: function subscribe(ob){...}, map: function map(transform){...}}
.map(val => val / 10)  //   返回新的observable对象 {subscribe: function subscribe(ob){next(val): y = transform(val) inputObservable.next(y), ...}, map: fn}
.filter(val => val !== 2)  
.subscribe(observe)    //新的observable对象里面subscribe

• subscribe
• filterObservable
• mapObservable
• arrayObservable
• arrayObservable.bsubscribe(ob)
• ob.next
• mapObserve.next()
• filterObserve.next()
• arrayfilterObserve.next()