Rxjs simple implementation

rxjs-simple-implementation

function map(transformFn) {
    const inputObservable = this
    const outputObservable = createObservable(function subscribe(outputObserver) {
      inputObservable.subscribe({
        next: function nextCallback(data) {
          outputObserver.next(transformFn(data))
        },
        error: (err) => outputObserver.error(err),
        complete: () => outputObserver.complete()
      })
    })
    return outputObservable
}

function filter(conditionFn) {
  const inputObservable = this
  const outputObservable = createObservable(function subscribe(outputObserver) {
    inputObservable.subscribe({
      next: function nextCallback(data) {
        if (conditionFn(data)) {
          outputObserver.next(data)
        }
      },
      error: (err) => outputObserver.error(err),
      complete: () => outputObserver.complete()
    })
  })
  return outputObservable
}


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

const arrayObservable = createObservable(function subscribe(ob){
    [1,2,3,4].forEach(ob.next)
    ob.complete()
  })

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

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

const batteryObservable = createObservable(function subscribe(ob) {
  navigator.getBattery().then((battery) => battery.addEventListener('chargingchange', ob.next));
})

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

  error: function errorCallback(err) {
    console.error(err)
  },

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


batteryObservable
  .map(event => event.currentTarget.level)
  //.filter(x => x > 200)
  .subscribe(observer)