observables.js

// OPERATORS
function map(transformFn) {
  const inputObservable = this;
  const outputObservable = createObservable(function subscribe(outputObserver) {
    inputObservable.subscribe({
      next: function(x) {
        const y = transformFn(x);
        outputObserver.next(y)
      },
      error: function(e) {
        outputObserver.error(e)
      },
      complete: function() {
        outputObserver.complete()
      }
    })
  })
  return outputObservable
}

function delay(period) {
  const inputObservable = this;
  const outputObservable = createObservable(function subscribe(outputObserver) {
    inputObservable.subscribe({
      next: function(x) {
        setTimeout(function() {
          outputObserver.next(x)
        }, period)
      },
      error: function(e) {
        outputObserver.error(e)
      },
      complete: function() {
        outputObserver.complete()
      }
    })
  })
  return outputObservable
}

function filter(transformFn) {
  const inputObservable = this;
  const outputObservable = createObservable(function subscribe(observer) {
    inputObservable.subscribe({
      next: function(x) {
        if (transformFn(x)) {
          observer.next(x)
        }
      },
      error: function(e) {
        observer.error(e)
      },
      complete: function() {
        observer.complete();
      }
    })
  })
  
  return outputObservable;
}

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

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

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

const fetchObservable = createObservable(function (ob) {
    fetch('https://jsonplaceholder.typicode.com/posts')
      .then(ob.next, ob.error)
      .finally(ob.complete)
  })

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

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

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

// SUBSCRIPTION
fetchObservable.subscribe(observer);
clickObservable.map(x => x.clientX).delay(5000).subscribe(observer);
arrayObservable
  .map(x => x/10)
  .filter(x => x !== 2)
  .subscribe(observer);