Hands on RxJS

rxjs.js

// https://npmcdn.com/@reactivex/[email protected]/dist/global/Rx.umd.js


Rx.Observable.create(function(observer) {
  let counter = 0;
  setInterval(function() {
    observer.next((counter += 2));
  }, 2000);
}).subscribe(
  function(val) {
    console.log("Next:- ", val);
  },
  function(val) {
    console.log("Error:- ", val);
  },
  function(val) {
    console.log("Completed:- ", val);
  }
);

Rx.Observable.of(2, 5, 9).subscribe(
  function(val) {
    console.log(`Next : ${val}`);
  },
  function() {},
  function(val) {
    console.log(`Completed : ${val}`);
  }
);

Rx.Observable.fromArray([2, 5, 9]).subscribe(
  function(val) {
    console.log(`Next : ${val}`);
  },
  function() {},
  function(val) {
    console.log(`Completed : ${val}`);
  }
);

Rx.Observable.fromEvent(document, "click").subscribe(
  function(val) {
    console.log(`Next : ${val}`);
  },
  function() {},
  function(val) {
    console.log(`Completed : ${val}`);
  }
);

Rx.Observable.range(29, 34).subscribe(
  function(val) {
    console.log(`Next : ${val}`);
  },
  function() {},
  function(val) {
    console.log(`Completed : ${val}`);
  }
);

const observable = Rx.Observable.range(1, 10);
const filtered = observable.filter(a => {
  return a % 2 === 0;
});

filtered.subscribe(function(val) {
  console.log(`Next : ${val}`);
});

const observable = Rx.Observable.range(26, 10);

observable
  .first(a => {
    return a % 4 === 0;
  })
  .subscribe(function(val) {
    console.log(`Next : ${val}`);
  });

const observable = Rx.Observable.range(26, 10);

observable
  .last(a => {
    return a % 4 === 0;
  })
  .subscribe(function(val) {
    console.log(`Next : ${val}`);
  });

const observable = Rx.Observable.from([1, 1, 2, 2, 3, 1, 2, 3]);

observable
  .distinctUntilChanged(function(o, n) {
    return o === n;
  })
  .subscribe(function(val) {
    console.log(`Next: ${val}`);
  });

const observable = Rx.Observable.timer(500, 1000);

observable.take(3).subscribe(val => console.log(val));
observable
  .take(3)
  .toArray()
  .subscribe(val => console.log(val));

const observable = Rx.Observable.create(observer => {
  console.log("Starts here ..");
  observer.next(1);
  observer.error(2);
  observer.error(4);
  observer.error(8);
  observer.next(10);
  observer.complete();
});

observable.retry(2).subscribe(val => console.log(val));

const observable = Rx.Observable.fromEvent(document, "click");

observable
  .debounce(() => Rx.Observable.timer(1000))
  .subscribe(val => console.log(val.timeStamp));

const observable = Rx.Observable.fromEvent(document, "click");

observable
  .throttle(() => Rx.Observable.timer(1000))
  .subscribe(val => console.log(val.timeStamp));

const observable = Rx.Observable.from([1, 3, 7]);

observable.every(val => val % 2).subscribe(val => console.log(val));

const observable = Rx.Observable.from([1, 3, 7]);

observable.map(val => val * 2).subscribe(val => console.log(val));

const intervalObservable = Rx.Observable.interval(1000);
const bufferObservable = Rx.Observable.fromEvent(document, "click");

intervalObservable.buffer(bufferObservable).subscribe(val => console.log(val));

const intervalObservable = Rx.Observable.interval(1000);
const bufferObservable = Rx.Observable.timer(3000, 2000);

intervalObservable.buffer(bufferObservable).subscribe(val => console.log(val));

const observable = Rx.Observable.interval(1000);
observable.bufferCount(4).subscribe(val => console.log(val));

const observable = Rx.Observable.interval(1000);
observable.bufferCount(4, 1).subscribe(val => console.log(val));

const intervalObservable = Rx.Observable.interval(1000);
const eventObservable = Rx.Observable.fromEvent(document, "click");

intervalObservable
  .concat(eventObservable)
  .subscribe(val => console.log(val.timeStamp || val));

const source = Rx.Observable.interval(1000);

source
  .take(1)
  .mergeMap(val => Rx.Observable.interval(3000).take(2))
  .subscribe(val => {
    console.log("begins");
    console.log(val);
    console.log("ends");
  });

const observable = Rx.Observable.from([
  { name: "Sue", age: 25 },
  { name: "Joe", age: 30 },
  { name: "Frank", age: 25 },
  { name: "Sarah", age: 35 }
]);

observable
  .groupBy(val => val.age)
  .mergeMap(
    val => val.toArray(),
    (x, y) => {
      console.log(y);
    }
  )
  .subscribe();

Rx.Observable.timer(1000)
  .concat(Rx.Observable.timer(1000))
  .subscribe(val => console.log(val));

Rx.Observable.timer(1000)
  .mapTo("FIRST!")
  .race(Rx.Observable.timer(200).mapTo("SECOND!"))
  .subscribe(val => console.log(val));

Rx.Observable.from([1, 3, 10])
  .reduce((acc, curr) => acc + curr)
  .subscribe(val => console.log(val));
Rx.Observable.from([1, 3, 10])
  .scan((acc, curr) => acc + curr)
  .subscribe(val => console.log(val));

Rx.Observable.of("World")
  .startWith("Hello")
  .subscribe(val => console.log(val));

Rx.Observable.zip(
  Rx.Observable.of("Hello").delay(4000),
  Rx.Observable.of("Birdy!")
).subscribe(val => console.log(val));
Rx.Observable.merge(
  Rx.Observable.of("Hello").delay(4000),
  Rx.Observable.of("Birdy!")
).subscribe(val => console.log(val));

Rx.Observable.from([
  { name: "Sue", age: 25 },
  { name: "Joe", age: 30 },
  { name: "Frank", age: 25 },
  { name: "Sarah", age: 35 }
])
  .pluck("age")
  .subscribe(val => console.log(val));

Rx.Observable.of("hello", "world", "!")
  .toArray()
  .repeat(3)
  .subscribe(val => console.log(val));

Rx.Observable.of("hello")
  .concat(Rx.Observable.of("world"))
  .subscribe(val => console.log(val));
Rx.Observable.of("hello", "goodbye")
  .concatMap(val => Rx.Observable.of(`${val} World!`).delay(2000))
  .subscribe(val => console.log(val));
Rx.Observable.of("hello", "goodbye")
  .mergeMap(val => Rx.Observable.of(`${val} World!`).delay(2000))
  .subscribe(val => console.log(val));

Rx.Observable.interval(1000)
  .take(3)
  .concatMapTo(Rx.Observable.of("FooBar").delay(2000))
  .subscribe(val => console.log(val));
Rx.Observable.interval(1000)
  .take(3)
  .mergeMapTo(Rx.Observable.of("FooBar").delay(2000))
  .subscribe(val => console.log(val));

Rx.Observable.of(
  Rx.Observable.interval(1000).take(5),
  Rx.Observable.interval(400).take(2),
  Rx.Observable.interval(2000).take(1)
)
  .concatAll()
  .subscribe(val => console.log(val));

Rx.Observable.of(
  Rx.Observable.interval(1000).take(5),
  Rx.Observable.interval(400).take(2),
  Rx.Observable.interval(2000).take(1)
)
  .mergeAll()
  .subscribe(val => console.log(val));

const [one, two] = Rx.Observable.from([3, 7, 10, 12, 27, 6]).partition(
  val => val <= 10
);

Rx.Observable.merge(
  one.map(val => `less than 10 : ${val}`),
  two.map(val => `more than 10 : ${val}`)
).subscribe(val => console.log(val));

Rx.Observable.of(4)
  .expand(val => Rx.Observable.of(val + 1))
  .take(3)
  .subscribe(val => console.log(val));