integration test geteventstore using rxjs has race condition

Question

Sorry, this one is a bit messy. My project is in nodejs. I have a test in mocha. In it I open a connection to geteventstore and subscribe to a stream. This essentially starts emitting events.

I wrap that event subscription in an rxjs observable and then write it to the console.

half of the time I get a stream full of events half of the time I don't.

I get the sense that the eventloop starts listening, doesn't hear anything and closes before the geteventstore can start blasting it with events.

I'm at a bit of a loss. I can tell the geteventstore is sending data cuz half the time I get it. My understanding is that as long as there is an someone is subscribed to an event, e.g. there is an eventlistener, the loop will stay open.

So perhaps the problem is with rxjs?

I don't know, any help would be greatly appreciated.

----EDIT

I don't know if this will help but the test looks like this.

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

so the mut is a connection to geteventstore, rx is rxjs, and the subscription object is an event emmiter that pumps data out of the geteventstore.

I understand that the problem is conflated by the fact that it deals wit at least two somewhat unusual products, the geteventstore, and the rxjs.

I mean I"m pretty confident that the gesConnection and subscription are, in fact, connecting and emitting. I just don't know how to test/investigate further.

Thanks

Answer 1

I don't see you making use of Mocha's async testing facilities.

MochaJs does not know that it should wait around for your test longer than it takes for your function to return.

Usually you'd return a promise:

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