How to filter RXJS ArrayObservable?

Question

I have an array which I'd like to filter.. sounds simple I know. But when I do, I still get the entire Array...

constructor(http:Http) {
    this._val = Math.random();
    let s = http.get('https://secure.digitalsignage.com/Digg');
    s.map(s => {
        let news = JSON.parse(s._body);
        return Rx.Observable.fromArray(news);
    }).filter(function(data) {
        console.log('all array ' + data);
        return true;
    }).subscribe(function (v) {
        console.log(v);
    });
}

so in console.log('all array ' + data); I am getting the entire array instead of a stream of individual array members, why?

here is debug snap:

I know I am not crazy because this works as expected:

 Rx.Observable.fromArray([1, 2, 3, 4, 5]).filter(function (v) {
        if (v < 3)
            return true
        return false;
    }).subscribe(function (v) {
        console.log(v);
    })

what gives?

tx for reading,

Sean.

Answer 1

You need to use concatMap/flatMap or any other operator that flattens a sequence of observables.

As of now, .filter receives an observable rather than the individual elements of the array (all you are doing is turning the array into an observable of an array. Your second example works as you are applying filter to that observable directly).

Try this:

constructor(http:Http) {
  this._val = Math.random();
  let s = http.get('https://secure.digitalsignage.com/Digg');
  s.flatMap(s => {
    let news = JSON.parse(s._body);
    return Rx.Observable.fromArray(news);
  }).filter(function(data) {
    console.log('all array ' + data);
    return true;
  }).subscribe(function (v) {
    console.log(v);
  });
}

You need to do this, as you would otherwise (using #map) have an observable of observables (you return an observable inside the observable) but want to work on the actual values contained within news (not on the observable containing them). It's somewhat like the difference between concatenating an array and just adding it as an element.

Another valid way of doing this would be using map like you did before but merging the emitted observables (flatMap/concatMap just map & merge in one go)

constructor(http:Http) {
  this._val = Math.random();
  let s = http.get('https://secure.digitalsignage.com/Digg');
  s.map(s => {
    let news = JSON.parse(s._body);
    return Rx.Observable.fromArray(news);
  }).mergeAll() // here you flatten the observable - i.e. you emit just the values contained in incoming observables and lose the observables themselves
    .filter(function(data) {
    console.log('all array ' + data);
    return true;
  }).subscribe(function (v) {
    console.log(v);
  });
}

If you didn't understand any of this don't worry, there's people who can explain it better than me :)