Question

33

RxJs: How to loop based on state of the observable?

rated 0 times [ 35] [ 2] / answers: 1 / hits: 18582 / 9 Years ago, sat, december 12, 2015, 12:00:00

I'm trying to get RxJs to loop over an Observable in my stream until it is in a certain state, then have the stream continue. Specifically I'm converting a synchronous do/while loop to RxJs, but I assume the same answer could be used for a for or while loop as well.

I thought I could use doWhile() for this, but it seems like the condition function does not have access to the item in the stream, which seems to defeat the purpose to me.

I'm not completely sure what the correct reactive terminology is for what I want, but here is an example of what I am going for:

var source = new Rx.Observable.of({val: 0, counter: 3});



source.map(o => {

  o.counter--;

  console.log('Counter: ' + o.counter);



  if (!o.counter) {

    o.val = YESS!;

  }

  return o;

})

.doWhile(o => { 

  return o.counter > 0; 

})

.subscribe(

    function (x) {

        console.log('Next: ' + x.val);

    },

    function (err) {

        console.log('Error: ' + err);   

    },

    function () {

        console.log('Completed');   

    });

The expected output would be:

Counter: 3

Counter: 2

Counter: 1

Counter: 0

Next: YESS!

Completed

Assuming this is a solvable problem, I am unclear on how you mark the 'start' of where you want to return when you loop.

Answers

Only authorized users can answer the question. Please sign in first, or register a free account.

nikoguym

Add To Favorites

Follow

Total Points: 339

Total Questions: 106

Total Answers: 95

Location: Mali

Member since Sat, Feb 12, 2022

2 Years ago

answered 9 Years ago terrence · Accepted Answer

There is the expand operator which gets you close by allowing you to recursively call a selector function. Returning an empty observable would be your break in that case. See jsbin:

var source = Rx.Observable.return({val: 0, counter: 3})

    .expand(value =>  { 

      if(!value.counter) return Rx.Observable.empty();

      value.counter -= 1;

      if(!value.counter) value.val = 'YESS';

      return Rx.Observable.return(value)

    })

    .subscribe(value => console.log(value.counter ? 

                                    'Counter: ' + value.counter : 

                                    'Next: ' + value.val));