when using Observable.fromEvent() bufferTime() take(3) and subscribe() - error "Cannot read property 'splice' of null" after last event is emitted

Question

var source = Observable.fromEvent(document.body, 'keypress');


var obs = source

  .bufferTime(1000)

  .map((clickBuffer) => {

    console.log(clickBuffer)

    return clickBuffer.length;
  })

  .take(3)

;

// when using this function - error after 3rd event
obs.subscribe((num) => {

});

I am trying this on angular 2 beta and RxJS 5. It gets 3 events with no problems and calculates the keypresses per second. Why do I get the error? I want it to stop after 3 seconds.

This is the continuation of this question: Counting keypresses per second with angular 2 Rxjs

Update

Found the case how make not throw, defining source different way:

// there are functions in typescript class
// onKeypressReactBuffer is called from angular template on input field keypress
class K {

    onKeypressReactBuffer() {}


    reactiveWay() {

    const source = Observable.create(observer => {

        // next(1) - we will use this value to add
        this.onKeypressReactBuffer = () => { 
            observer.next(1);      


            // this make it not throw the error. 
            observer.complete('ha') };
        });

    }
}

So still question remains - why it does not work with function fromEvent() ? I do not see how to define completed() method on this function also. So it should be some default.

And also - how could I find faster that for Observable.create I needed .complete() from the information the error message gave me? It only recently after hours of thinking popped to my head to try - maybe it needs completed() function.

Update

Actually my last updated code does not work correctly. It just stops giving me error.

What happens with last update is - it emits events of 0 and 3 times if not pressing key. Once pressing - it emits event of sum 1 and stops emitting.

Update:

How to reproduce on web pack starter:

Install angular 2 webpack starter version 5.0.3 (with version 3 also I got the error).

Btw I had to change in package.json from Rxjs beta 4 to beta 2, because it was failing to install otherwise.

And in home.comoponent.ts ngOnInit() function put the code almost same as https://jsbin.com/fafuladayi/edit?js,console,output

only difference is that instead of Rx.Observable use Observable and import the Observable - see the code:

import {Component} from 'angular2/core';
import {AppState} from '../app.service';

import {Title} from './title';
import {XLarge} from './x-large';

import { Observable, Subject } from 'rxjs/Rx';

@Component({
  // The selector is what angular internally uses
  // for `document.querySelectorAll(selector)` in our index.html
  // where, in this case, selector is the string 'home'
  selector: 'home',  // <home></home>
  // We need to tell Angular's Dependency Injection which providers are in our app.
  providers: [
    Title
  ],
  // We need to tell Angular's compiler which directives are in our template.
  // Doing so will allow Angular to attach our behavior to an element
  directives: [
    XLarge
  ],
  // We need to tell Angular's compiler which custom pipes are in our template.
  pipes: [ ],
  // Our list of styles in our component. We may add more to compose many styles together
  styles: [ require('./home.css') ],
  // Every Angular template is first compiled by the browser before Angular runs it's compiler
  template: require('./home.html')
})
export class Home {
  // Set our default values
  localState = { value: '' };
  // TypeScript public modifiers
  constructor(public appState: AppState, public title: Title) {

  }

  ngOnInit() {
    console.log('hello `Home` component');
    // this.title.getData().subscribe(data => this.data = data);


    var source = Observable.fromEvent(document.body, 'keypress');


    var obs = source

        .bufferTime(1000)

        .map((clickBuffer) => {

          //console.log(clickBuffer.length)

          return clickBuffer.length;
        })

        .take(5)

        ;

    // when using this function - error after 3rd event
    obs.subscribe((num) => {

      console.log(' home ' + num)

    });


  }

  submitState(value) {
    console.log('submitState', value);
    this.appState.set('value', value);
  }

}

And Thierry Templier example works, but I want to understand also why does not work the Birowsky example with this starter pack.

Btw I also see that Birowsky uses older version - rxjs@5.0.0-alpha.8 . I tried to include beta.2 with this url, but then I get error Rx not found: https://npmcdn.com/@reactivex/rxjs@5.0.0-beta.2/dist/global/Rx.js in jsbin.com

Answer 1

I think that you could use a dedicated subject to provide the takeUntil operator to stop data flow. Something like that:

var source = Observable.fromEvent(document.body, 'keyup');

var stop = new Subject();
Observable.interval(3100).subscribe(() => stop.next());

var obs = source.bufferTime(1000).map(clickBuffer => {
  console.log(clickBuffer);
  return clickBuffer.length;
}).takeUntil(stop);

obs.subscribe((num) => {
  console.log('num = '+num);
});