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.
Not exactly what you want but close, using expand
operator, and signalling end of recursion with Rx.Observable.empty
(http://jsfiddle.net/naaycu71/3/):
var source = new Rx.Observable.of({val: 0, counter: 3});
source.expand(function(o) {
console.log('Counter: ' + o.counter);
o.counter--;
return (o.counter >= 0) ? Rx.Observable.just(o) : Rx.Observable.empty()
})
.subscribe(
function (x) {
console.log('Next: ' , x);
},
function (err) {
console.log('Error: ' + err);
},
function () {
console.log('Completed');
});
Output :
Next: Object {val: 0, counter: 3}
Counter: 3
Next: Object {val: 0, counter: 2}
Counter: 2
Next: Object {val: 0, counter: 1}
Counter: 1
Next: Object {val: 0, counter: 0}
Counter: 0
Completed
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));
If you love us? You can donate to us via Paypal or buy me a coffee so we can maintain and grow! Thank you!
Donate Us With