node.js - 使用rxjs的集成测试geteventstore具有竞争条件

抱歉，这有点混乱。我的项目在nodejs中。我正在摩卡咖啡中进行测试。在其中，我打开到geteventstore的连接并订阅流。这基本上开始发出事件。

我将该事件订阅包装在可观察的rxjs中，然后将其写入控制台。

一半的时间，我有一半的时间都没有事件流。

我感觉到eventloop开始侦听，什么也没听到，然后在geteventstore开始对事件进行爆炸之前关闭。

我有点茫然。我可以知道geteventstore在一半时间内发送数据cuz。我的理解是，只要有人订阅了某项活动，例如有一个事件监听器，循环将保持打开状态。

那么也许问题出在rxjs上？

我不知道，我们将不胜感激。

- - 编辑

我不知道这是否有帮助，但是测试看起来像这样。

context('when calling subscription', ()=> {
    it('should stay open', function () {
        mut = bootstrap.getInstanceOf('gesConnection');
        var rx = bootstrap.getInstanceOf('rx');
        var subscription = mut.subscribeToAllFrom();

        rx.Observable.fromEvent(subscription, 'event').forEach(x=> console.log(x));

        subscription.on('event', function (payload) {
            console.log('event received by dispatcher');
            console.log('event processed by dispatcher');
        });
        mut._handler._connectingPhase.must.equal('Connected');
    })
});

因此mut是到geteventstore的连接，rx是rxjs，订阅对象是事件发射器，它将数据从geteventstore中抽出。

我了解到，该问题与以下事实混为一谈：它至少处理了两种不太寻常的产品，即geteventstore和rxjs。

我的意思是，我非常有信心gesConnection和subscription实际上是在连接和发出信号。我只是不知道如何进行进一步的测试/研究。

谢谢

最佳答案

我看不到您使用Mocha's async testing facilities。

MochaJs不知道它应该比函数返回所需的等待时间更长。

通常您会返回一个承诺：

    it('must stay open', () => {
        mut = bootstrap.getInstanceOf('gesConnection');
        var rx = bootstrap.getInstanceOf('rx');
        var subscription = mut.subscribeToAllFrom();

        subscription.on('event', function (payload) {
            console.log('event received by dispatcher');
            console.log('event processed by dispatcher');
        });

        var promise = rx.Observable
            .fromEvent(subscription, 'event')
            .take(100) // stop test after 100 events
            .do(x => console.log(x))
            .finally(() => {
                // do any cleanup here.
                // such as close your connection
                // or "subscription" variable
            })
            .toPromise();

        mut._handler._connectingPhase.must.equal('Connected');

        // tells Mocha to wait until the observable completes
        return promise;
    });