takeLast seems to kill observable stream

Question

I'm trying to setup an observable stream where I compare the most recent set value to the last one, and I'd like to filter based on some changes between the two. I setup a BehaviorSubject in a service (Angular 2, if that matters) and created function to return it:

getFilters() {
    return this.filtersSubject.asObservable();
}

Then in the component using it, I'm trying this:

this.eventFilterService.getFilters()
    .takeLast(2)
    .subscribe((data) => console.log(data));

But I get no console logs. If I remove the takeLast, I see my data returning. From the docs, my assumption would be as is, it would just fire twice. My goal was to feed the takeLast into a reduce then a filter.

Am I using takeLast wrong?

Answer 1

takeLast(x) will wait for completion of the Observable before it can emit the last x emissions. Since you use a subject you need to invoke complete() for it to work.

But you really want to calculate a difference between two emission. To do so use the .buffer() operator

Rx.Observable.from([2,3,5,8])
     .bufferCount(2,1)
     .map(buff =>  buff[1] - buff[0])
     .subscribe(delta => console.log(delta))