node.js - integration test geteventstore using rxjs has race condition -

sorry, 1 bit messy. project in nodejs. have test in mocha. in open connection geteventstore , subscribe stream. starts emitting events.

i wrap event subscription in rxjs observable , write console.

half of time stream full of events half of time don't.

i sense eventloop starts listening, doesn't hear , closes before geteventstore can start blasting events.

i'm @ bit of loss. can tell geteventstore sending data cuz half time it. understanding long there subscribed event, e.g. there eventlistener, loop stay open.

so perhaps problem rxjs?

i don't know, appreciated.

----edit

i don't know if test looks 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 dispatcher');             console.log('event processed dispatcher');         });         mut._handler._connectingphase.must.equal('connected');     }) }); 

so mut connection geteventstore, rx rxjs, , subscription object event emmiter pumps data out of geteventstore.

i understand problem conflated fact deals wit @ least 2 unusual products, geteventstore, , rxjs.

i mean i"m pretty confident gesconnection , subscription are, in fact, connecting , emitting. don't know how test/investigate further.

thanks

i don't see making use of mocha's async testing facilities.

mochajs not know should wait around test longer takes function return.

usually you'd return 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 dispatcher');             console.log('event processed dispatcher');         });          var promise = rx.observable             .fromevent(subscription, 'event')             .take(100) // stop test after 100 events             .do(x => console.log(x))             .finally(() => {                 // cleanup here.                 // such close connection                 // or "subscription" variable             })             .topromise();          mut._handler._connectingphase.must.equal('connected');          // tells mocha wait until observable completes         return promise;     });