eliminate race condition properly

Author: davidjgoss

package-lock.json

@@ -43,6 +43,7 @@
     "@cucumber/query": "^14.0.0",
     "@cucumber/tag-expressions": "8.1.0",
     "globby": "^16.0.0",
+    "rxjs": "7.8.2",
     "stack-utils": "2.0.6"
   },
   "devDependencies": {

package.json

@@ -1,7 +1,4 @@
-import { EventEmitter } from 'node:events'
-
 import { Envelope } from '@cucumber/messages'
+import { ReplaySubject } from 'rxjs'
 
-export const eventEmitter = new EventEmitter<{
-  envelope: [Envelope]
-}>()
+export const envelopes$ = new ReplaySubject<Envelope>()

src/messages/eventEmitter.ts

@@ -5,7 +5,7 @@ import { Envelope } from '@cucumber/messages'
 import { Buffer } from 'buffer'
 
 import { deriveSocketPath } from './deriveSocketPath.js'
-import { eventEmitter } from './eventEmitter.js'
+import { envelopes$ } from './eventEmitter.js'
 
 export async function setupMessageListening() {
   const pid = process.pid.toString()
@@ -50,7 +50,7 @@ class Deframer {
     for (const line of lines) {
       if (line) {
         const envelope = JSON.parse(line.toString()) as Envelope
-        eventEmitter.emit('envelope', envelope)
+        envelopes$.next(envelope)
       }
     }
   }

src/messages/setupMessageListening.ts

@@ -1,7 +1,7 @@
 import { connect } from 'node:net'
 
 import { deriveSocketPath } from './deriveSocketPath.js'
-import { eventEmitter } from './eventEmitter.js'
+import { envelopes$ } from './eventEmitter.js'
 
 export async function setupMessageWriting() {
   const listenerPid = process.env.CUCUMBER_MESSAGES_LISTENING
@@ -12,7 +12,7 @@ export async function setupMessageWriting() {
       const socketPath = deriveSocketPath(listenerPid)
       const client = connect(socketPath, resolve)
 
-      eventEmitter.on('envelope', (envelope) => {
+      envelopes$.subscribe((envelope) => {
         client.write(JSON.stringify(envelope) + '\n')
       })
 

src/messages/setupMessageWriting.ts

@@ -10,14 +10,12 @@ import {
 } from '@cucumber/messages'
 
 import { makeTimestamp } from '../makeTimestamp.js'
-import { eventEmitter, setupMessageListening } from '../messages/index.js'
+import { envelopes$, setupMessageListening } from '../messages/index.js'
 import { newId } from '../newId.js'
 import { mapTestStepResult } from './mapTestStepResult.js'
 import { meta } from './meta.js'
 
 await setupMessageListening()
-const rawEnvelopes: Array<Envelope> = []
-eventEmitter.on('envelope', (envelope) => rawEnvelopes.push(envelope))
 
 export async function* generateEnvelopes(
   source: AsyncIterable<TestEvent>
@@ -26,8 +24,11 @@ export async function* generateEnvelopes(
 
   const nodeFailOrPassEvents: Array<EventData.TestFail | EventData.TestPass> = []
   const testStepFinishedMessages: Array<TestStepFinished> = []
+  const rawEnvelopes: Array<Envelope> = []
   const testRunEnvelopes: Array<Envelope> = []
 
+  envelopes$.subscribe((envelope) => rawEnvelopes.push(envelope))
+
   const testRunStarted: TestRunStarted = {
     id: newId(),
     timestamp: makeTimestamp(),

src/reporters/generateEnvelopes.ts

@@ -1,9 +1,9 @@
 import { Envelope } from '@cucumber/messages'
 
-import { eventEmitter } from '../messages/index.js'
+import { envelopes$ } from '../messages/index.js'
 
 export class MessagesCollector {
   push(...envelopes: ReadonlyArray<Envelope>) {
-    envelopes.forEach((envelope) => eventEmitter.emit('envelope', envelope))
+    envelopes.forEach((envelope) => envelopes$.next(envelope))
   }
 }