RxJS：在 NodeJS 中处理 groupBy 和 Observable.fromEvents

Question

我对 RxJS 印象深刻，并开始着手研究。但是，我的以下 nodejs 代码至少对我来说不能按预期工作。

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' } );

我希望subscribe 产生一些输出

Answer 1

这是我在 JSBin http://jsbin.com/cufana/edit?js,console 中创建的稍有改动的版本。

我在您的代码中看到的问题是您有一个未完成的可观察对象。如果您正在执行 groupBy，则只要源 observable 未完成，内部保存的结果（即分组结果）就不会被推送。

let events = new Rx.Subject();

events
    .groupBy( event => event.type)
    .flatMap(group => group.reduce((acc, curr) => [...acc, curr], []))
    .subscribe( ( data ) => {
        console.log( data );
    } );


events.next({ 'type': 1, msg: 'Test 1' } );
events.next({ 'type': 1, msg: 'Test 2' } );
events.next({ 'type': 2, msg: 'Test 3' } );
events.complete();

在这里你可以看到我已经将 eventemitter 更改为一个主题以获得一个没有 angular2 依赖的工作 jsbin。我正在完成这个主题，所以我可以从 groupBy 观察到的来源已经完成。这将推动结果通过。

其余代码非常正确。

如果 EventEmitter 确实来自 Angular2，我猜你在完成这个时会遇到问题。这可以从子组件中完成吗？

Answer 2

正如其他人所建议的，Observable 永远不会完成，因此任何需要前面的 Observable 完成的运算符都不会发出任何东西：

groupBy() 运算符会为每个组发出一个GroupedObservable 实例，以便您订阅它。我知道这会产生与您预期不同的结果，但也许您可以使用它：

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' } );

每个组都有自己的GroupedObservable。

这会打印到控制台：

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

Answer 3

问题是您的序列没有终止。 groupBy 仅在终止时触发。见jsbin example with both cases。

当您执行Rx.Observable.from(...) 时，您将获得一个完整的序列，否则，您需要手动终止。 groupBy documentation 上的示例证明了这一点。