CODEWITHSUNDEEP
javascriptpromiserxjsobservablerxjs5
Liang Zhou
Liang Zhou

Reputation: 2165

Combine a list of ordered operations into one Observable

var state = [];
var operation1 = function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state.push(1, 2);
        setTimeout(resolve, 300, state);
    }));
};
var operation2 = function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.map(x => x * 2);
        setTimeout(resolve, 200, state);
    }));
};
var operation3 = function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.reduce( (prev, next) => prev + next );
        setTimeout(resolve, 100, state);
    }));
};
var operations = [operation1, operation2, operation3];

Given the code above, I am trying to combine operations into one Observable that emits the state of each operation. So the Observable needs to do either one of the following:

Upvotes: 2

Views: 109

Answers (2)

user3743222
user3743222

Reputation: 18665

You could try (jsbin)

var state = [];
var operation1 = Rx.Observable.defer(function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state.push(1, 2);
        setTimeout(resolve, 300, state);
    }));
});
var operation2 = Rx.Observable.defer(function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.map(x => x * 2);
        setTimeout(resolve, 200, state);
    }));
});
var operation3 = Rx.Observable.defer(function() {
    return Rx.Observable.fromPromise(new Promise((resolve, reject) => {
        state = state.reduce( (prev, next) => prev + next );
        setTimeout(resolve, 100, state);
    }));
});
var operations = Rx.Observable.from([operation1, operation2, operation3]).merge(1);
operations.subscribe(function(x){console.log(x)})

Please check if that does the trick, I will elaborate later on how this works.

Upvotes: 3

denixtry
denixtry

Reputation: 3098

You can use Rx.Observable.concat but I think this simple case would be easier without rx.java and using Promise.all

let slow = new Promise((resolve) => {
  setTimeout(resolve, 200, 'slow');
});
let instant = new Promise((resolve) => {
  setTimeout(resolve, 0, 'instant');
});
let quick = new Promise((resolve) => {
  setTimeout(resolve, 50, 'quick');
});

var operation1 = function() {
  return Rx.Observable.fromPromise(slow);
}
var operation2 = function() {
  return Rx.Observable.fromPromise(instant);
}
var operation3 = function() {
  return Rx.Observable.fromPromise(quick);
}
var operations = [operation1(), operation2(), operation3()];
var source = Rx.Observable.concat(operations);

var subscription = source.subscribe(
  function(x) {
    console.log('Next: ' + x);
  },
  function(err) {
    console.log('Error: ' + err);
  },
  function(a) {
    console.log('Completed', a);
  });

// Or with promises

operations = [slow, instant, quick]

Promise.all(operations)
.then(console.log.bind(console, "Promise all"))
<script src="https://cdnjs.cloudflare.com/ajax/libs/rxjs/4.1.0/rx.all.min.js"></script>

Upvotes: 0

Related Questions