observable

rx.js

const observer = {
  next: function nextCb(data) {
    console.log(data)
  },

  error: function errorCb(error) {
    console.error(error)
  },

  complete: function completeCb(data) {
    console.log('done')
  }
}

function map(transformFn) {
  const inputObservalbe = this
  const outputObservable = createObservable(function subscribe(outputObserver) {
    inputObservalbe.subscribe({
      next: x => {
        const y = transformFn(x)
        outputObserver.next(y)
      }
    })
  })
  return outputObservable
}

function filter(conditionFn) {
  const inputObservalbe = this
  const outputObservable = createObservable(function subscribe(outputObserver) {
    inputObservalbe.subscribe({
      next: x => {
        // const y = transformFn(x)
        if (conditionFn(x)) {
          outputObserver.next(x)          
        }
      }
    })
  })
  return outputObservable
}

function delay(period) {
  const inputObservalbe = this
  const outputObservable = createObservable(function subscribe(outputObserver) {
    inputObservalbe.subscribe({
      next: x => {
        setTimeout(() => {
          outputObserver.next(x)
        }, period)
      }
    })
  })
  return outputObservable
}

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

const arrayObservable = createObservable(function subscribe(ob) {
  console.log('hahaha')
  [1, 2, 3].forEach(ob.next)
  // ob.complete()
})

const clickObservable = createObservable(function subscribe(ob) {
  document.addEventListener('click', ob.next)
})

clickObservable
  .map(ev => ev.clientX)
  .filter(x => x < 200)
  .subscribe(observer)