node.js - RxJS: Working on groupBy and Observable.fromEvents in NodeJS -

i'm impressed rxjs , start working on that. however, following nodejs code not work expected @ least me.

let events = new eventemitter(); let source = rx.observable.fromevent( events, 'data' );  source     .groupby( event => event.type )     .flatmap( group => group.reduce( ( acc, cur ) => _.merge( acc, cur ), [] ) )     .subscribe( ( data ) => {         console.log( data );     } );   events.emit( 'data', { 'type': 1, msg: 'test 1' } ); events.emit( 'data', { 'type': 1, msg: 'test 2' } ); events.emit( 'data', { 'type': 2, msg: 'test 3' } ); 

i expect subscribe produces output

as suggested others observable never completes operator requires preceding observable complete never emit anything:

the groupby() operator emits instance of groupedobservable each group can subscribe it. know produces different result you'd expected maybe can work it:

const rx = require('rxjs/rx'); const eventemitter = require('events');  let events = new eventemitter(); let source = rx.observable.fromevent(events, 'data');  source     .groupby( event => event.type )     .subscribe( ( groupedobservable ) => {         groupedobservable.subscribe(val => {             console.log(groupedobservable.key, val);         });     } );   events.emit( 'data', { 'type': 1, msg: 'test 1' } ); events.emit( 'data', { 'type': 1, msg: 'test 2' } ); events.emit( 'data', { 'type': 5, msg: 'test 3' } ); 

each group has it's own groupedobservable.

this prints console:

1 { type: 1, msg: 'test 1' } 1 { type: 1, msg: 'test 2' } 5 { type: 5, msg: 'test 3' }