André Staltz (@andrestaltz): You will learn RxJS at ng-europe 2016

rx.js

/*
 * André Staltz (@andrestaltz): You will learn RxJS at ng-europe 2016
 * https://www.youtube.com/watch?v=uQ1zhJHclvs
 */

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: e => outputObserver.error(e),
      complete: () => outputObserver.complete()
    });
  });
  return outputObservable;
}

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

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

function createObservable(subscribe) {
  return {
    subscribe: subscribe, // "subscribe" === "giveMeSomeData"
    map: map,
    filter: filter,
    delay: delay
  };
}

// Using it:
// ---------

// "Observable" === "stream of events"
const arrayObservable = createObservable(function subscribe(ob) {
  [10, 20, 30].forEach(ob.next);
  ob.complete();
});

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

const observer = {
  next: function nextCallback(data) {
    console.log(data);
  },
  error: function errorCallback(data) {
    console.log(data);
  },
  complete: function completeCallback(data) {
    console.log("done");
  }
};

arrayObservable
  .map(x => x / 10)
  .filter(x => x !== 2)
  .subscribe(observer);

clickObservable
  .map(ev => ev.clientX)
  .filter(x => x > 200)
  .delay(2000)
  .subscribe(observer);