Improved tests resiliency around restarts

Author: oskardudycz

src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.handling.int.spec.ts

@@ -1,7 +1,9 @@
 import {
   assertThatArray,
+  asyncAwaiter,
   delay,
   getInMemoryDatabase,
+  JSONParser,
   type Event,
   type InMemoryReactorOptions,
   type RecordedMessage,
@@ -431,6 +433,7 @@ void describe('EventStoreDB event store started consumer', () => {
 
           const guestId = uuid();
           const streamName = `guestStay-${guestId}`;
+          const waitForStart = asyncAwaiter();
 
           // When
           const consumer = eventStoreDBEventStoreConsumer({
@@ -441,7 +444,8 @@ void describe('EventStoreDB event store started consumer', () => {
             processorId: uuid(),
             stopAfter: (event) =>
               event.metadata.globalPosition === stopAfterPosition,
-            eachMessage: (event) => {
+            eachMessage: async (event) => {
+              await waitForStart.wait;
               result.push(event);
             },
           });
@@ -459,6 +463,7 @@ void describe('EventStoreDB event store started consumer', () => {
               events,
             );
             stopAfterPosition = appendResult.lastEventGlobalPosition;
+            waitForStart.resolve();
 
             await consumerPromise;
 
@@ -492,6 +497,7 @@ void describe('EventStoreDB event store started consumer', () => {
 
           const result: GuestStayEvent[] = [];
           let stopAfterPosition: bigint | undefined = undefined;
+          const waitForStart = asyncAwaiter();
 
           // When
           const consumer = eventStoreDBEventStoreConsumer({
@@ -503,7 +509,8 @@ void describe('EventStoreDB event store started consumer', () => {
             startFrom: 'CURRENT',
             stopAfter: (event) =>
               event.metadata.globalPosition === stopAfterPosition,
-            eachMessage: (event) => {
+            eachMessage: async (event) => {
+              await waitForStart.wait;
               result.push(event);
             },
           });
@@ -516,6 +523,7 @@ void describe('EventStoreDB event store started consumer', () => {
               events,
             );
             stopAfterPosition = appendResult.lastEventGlobalPosition;
+            waitForStart.resolve();
 
             await consumerPromise;
 
@@ -555,6 +563,8 @@ void describe('EventStoreDB event store started consumer', () => {
           let result: GuestStayEvent[] = [];
           let stopAfterPosition: bigint | undefined = lastEventGlobalPosition;
 
+          const waitForStart = asyncAwaiter();
+
           // When
           const consumer = eventStoreDBEventStoreConsumer({
             connectionString,
@@ -565,26 +575,33 @@ void describe('EventStoreDB event store started consumer', () => {
             startFrom: 'CURRENT',
             stopAfter: (event) =>
               event.metadata.globalPosition === stopAfterPosition,
-            eachMessage: (event) => {
+            eachMessage: async (event) => {
+              await waitForStart.wait;
               result.push(event);
+              console.log(`Received event: ${JSONParser.stringify(event)}`);
             },
           });
 
-          await consumer.start();
+          let consumerPromise = consumer.start();
+          waitForStart.resolve();
+          await consumerPromise;
           await consumer.stop();
 
+          waitForStart.reset();
+
           result = [];
 
           stopAfterPosition = undefined;
 
           try {
-            const consumerPromise = consumer.start();
+            consumerPromise = consumer.start();
 
             const appendResult = await eventStore.appendToStream(
               streamName,
               events,
             );
             stopAfterPosition = appendResult.lastEventGlobalPosition;
+            waitForStart.resolve();
 
             await consumerPromise;
 
@@ -621,12 +638,14 @@ void describe('EventStoreDB event store started consumer', () => {
           let result: GuestStayEvent[] = [];
           let stopAfterPosition: bigint | undefined = lastEventGlobalPosition;
 
+          const waitForStart = asyncAwaiter();
           const processorOptions: InMemoryReactorOptions<GuestStayEvent> = {
             processorId: uuid(),
             startFrom: 'CURRENT',
             stopAfter: (event) =>
               event.metadata.globalPosition === stopAfterPosition,
-            eachMessage: (event) => {
+            eachMessage: async (event) => {
+              await waitForStart.wait;
               result.push(event);
             },
             connectionOptions: { database },
@@ -640,13 +659,15 @@ void describe('EventStoreDB event store started consumer', () => {
           try {
             consumer.reactor<GuestStayEvent>(processorOptions);
 
+            waitForStart.resolve();
             await consumer.start();
           } finally {
             await consumer.close();
           }
 
           result = [];
 
+          waitForStart.reset();
           stopAfterPosition = undefined;
 
           const newConsumer = eventStoreDBEventStoreConsumer({
@@ -662,6 +683,7 @@ void describe('EventStoreDB event store started consumer', () => {
               streamName,
               events,
             );
+            waitForStart.resolve();
             stopAfterPosition = appendResult.lastEventGlobalPosition;
 
             await consumerPromise;

src/packages/emmett/src/utils/promises.ts

@@ -1,3 +1,25 @@
 export const delay = (ms: number): Promise<void> => {
   return new Promise((resolve) => setTimeout(resolve, ms));
 };
+
+export type AsyncAwaiter<T = void> = {
+  wait: Promise<T>;
+  resolve: (value: T | PromiseLike<T>) => void;
+  // eslint-disable-next-line @typescript-eslint/no-explicit-any
+  reject: (reason?: any) => void;
+  reset: () => void;
+};
+
+// TODO: Remove this after migrating to Node 22
+export const asyncAwaiter = <T = void>(): AsyncAwaiter<T> => {
+  const result: AsyncAwaiter<T> = {} as AsyncAwaiter<T>;
+
+  (result.reset = () => {
+    result.wait = new Promise<T>((res, rej) => {
+      result.resolve = res;
+      result.reject = rej;
+    });
+  })();
+
+  return result;
+};