[javascript] Chaining Observables in RxJS

I'm learning RxJS and Angular 2. Let's say I have a promise chain with multiple async function calls which depend on the previous one's result which looks like:

var promiseChain = new Promise((resolve, reject) => {
  setTimeout(() => {
    resolve(1);
  }, 1000);
}).then((result) => {
  console.log(result);

  return new Promise((resolve, reject) => {
    setTimeout(() => {
      resolve(result + 2);
    }, 1000);
  });
}).then((result) => {
  console.log(result);

  return new Promise((resolve, reject) => {
      setTimeout(() => {
      resolve(result + 3);
        }, 1000);
  });
});

promiseChain.then((finalResult) => {
  console.log(finalResult);
});

My attempts at doing the same solely using RxJS without the use of promises produced the following:

var observableChain = Observable.create((observer) => {
  setTimeout(() => {
    observer.next(1);
    observer.complete();
  }, 1000);
}).flatMap((result) => {
  console.log(result);

  return Observable.create((observer) => {
    setTimeout(() => {
      observer.next(result + 2);
      observer.complete()
    }, 1000);
  });
}).flatMap((result) => {
  console.log(result);

  return Observable.create((observer) => {
    setTimeout(() => {
      observer.next(result + 3);
      observer.complete()
    }, 1000);
  });
});

observableChain.subscribe((finalResult) => {
  console.log(finalResult);
});

It yields the same output as the promise chain. My questions are

  1. Am I doing this right? Are there any RxJS related improvements that I can make to the above code

  2. How do I get this observable chain to execute repeatedly? i.e. Adding another subscription at the end just produces an additional 6 though I expect it to print 1, 3 and 6.

    observableChain.subscribe((finalResult) => { console.log(finalResult); });

    observableChain.subscribe((finalResult) => { console.log(finalResult); });

    1 3 6 6

This question is related to javascript promise reactive-programming rxjs rxjs5

The answer is


About promise composition vs. Rxjs, as this is a frequently asked question, you can refer to a number of previously asked questions on SO, among which :

Basically, flatMap is the equivalent of Promise.then.

For your second question, do you want to replay values already emitted, or do you want to process new values as they arrive? In the first case, check the publishReplay operator. In the second case, standard subscription is enough. However you might need to be aware of the cold. vs. hot dichotomy depending on your source (cf. Hot and Cold observables : are there 'hot' and 'cold' operators? for an illustrated explanation of the concept)


Examples related to javascript

need to add a class to an element How to make a variable accessible outside a function? Hide Signs that Meteor.js was Used How to create a showdown.js markdown extension Please help me convert this script to a simple image slider Highlight Anchor Links when user manually scrolls? Summing radio input values How to execute an action before close metro app WinJS javascript, for loop defines a dynamic variable name Getting all files in directory with ajax

Examples related to promise

Axios handling errors typescript: error TS2693: 'Promise' only refers to a type, but is being used as a value here Syntax for async arrow function Angular 2: How to call a function after get a response from subscribe http.post How to use fetch in typescript Returning Promises from Vuex actions Use async await with Array.map Getting a UnhandledPromiseRejectionWarning when testing using mocha/chai using setTimeout on promise chain Why is my asynchronous function returning Promise { <pending> } instead of a value?

Examples related to reactive-programming

Chaining Observables in RxJS How to get data from observable in angular2 Using onBlur with JSX and React What is (functional) reactive programming?

Examples related to rxjs

Angular - "has no exported member 'Observable'" What is pipe() function in Angular How to convert Observable<any> to array[] TypeError: You provided an invalid object where a stream was expected. You can provide an Observable, Promise, Array, or Iterable What is the difference between Subject and BehaviorSubject? Best way to import Observable from rxjs take(1) vs first() Observable Finally on Subscribe Getting an object array from an Angular service BehaviorSubject vs Observable?

Examples related to rxjs5

Observable Finally on Subscribe How to return value from function which has Observable subscription inside? Chaining Observables in RxJS