Fixed restarting stopped MongoDB Consumer

It appeared that when all processors was stopped then it was still
showing that's is running, as currently there's no notification from
processors (probably something to have eventually). For now, ensured
that flag is set to false when stopping.

Author: oskardudycz

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventStoreConsumer.inMemory.projections.int.spec.ts

@@ -300,79 +300,79 @@ void describe('mongoDB event store started consumer', () => {
       },
     );
 
-    // void it(
-    //   'handles only new events when CURRENT position is stored for restarted consumer',
-    //   withDeadline,
-    //   async () => {
-    //     // Given
-    //     const shoppingCartId = `shoppingCart:${uuid()}`;
-    //     const streamName = `shopping_cart-${shoppingCartId}`;
+    void it(
+      'handles only new events when CURRENT position is stored for restarted consumer',
+      withDeadline,
+      async () => {
+        // Given
+        const shoppingCartId = `shoppingCart:${uuid()}`;
+        const streamName = `shopping_cart-${shoppingCartId}`;
 
-    //     const initialEvents: ShoppingCartSummaryEvent[] = [
-    //       { type: 'ProductItemAdded', data: { productItem } },
-    //       { type: 'ProductItemAdded', data: { productItem } },
-    //     ];
-    //     const { nextExpectedStreamVersion } = await eventStore.appendToStream(
-    //       streamName,
-    //       initialEvents,
-    //     );
+        const initialEvents: ShoppingCartSummaryEvent[] = [
+          { type: 'ProductItemAdded', data: { productItem } },
+          { type: 'ProductItemAdded', data: { productItem } },
+        ];
+        const { nextExpectedStreamVersion } = await eventStore.appendToStream(
+          streamName,
+          initialEvents,
+        );
 
-    //     const events: ShoppingCartSummaryEvent[] = [
-    //       { type: 'ProductItemAdded', data: { productItem } },
-    //       {
-    //         type: 'ShoppingCartConfirmed',
-    //         data: { confirmedAt },
-    //       },
-    //     ];
+        const events: ShoppingCartSummaryEvent[] = [
+          { type: 'ProductItemAdded', data: { productItem } },
+          {
+            type: 'ShoppingCartConfirmed',
+            data: { confirmedAt },
+          },
+        ];
 
-    //     let stopAfterPosition: bigint | undefined = nextExpectedStreamVersion;
+        let stopAfterPosition: bigint | undefined = nextExpectedStreamVersion;
 
-    //     const inMemoryProcessor = inMemoryProjector<ShoppingCartSummaryEvent>({
-    //       processorId: uuid(),
-    //       projection: shoppingCartsSummaryProjection,
-    //       connectionOptions: { database },
-    //       startFrom: 'CURRENT',
-    //       stopAfter: (event) =>
-    //         event.metadata.streamName === streamName &&
-    //         event.metadata.streamPosition === stopAfterPosition,
-    //     });
+        const inMemoryProcessor = inMemoryProjector<ShoppingCartSummaryEvent>({
+          processorId: uuid(),
+          projection: shoppingCartsSummaryProjection,
+          connectionOptions: { database },
+          startFrom: 'CURRENT',
+          stopAfter: (event) =>
+            event.metadata.streamName === streamName &&
+            event.metadata.streamPosition === stopAfterPosition,
+        });
 
-    //     const consumer = mongoDBEventStoreConsumer<ShoppingCartSummaryEvent>({
-    //       connectionString,
-    //       clientOptions: { directConnection: true },
-    //       processors: [inMemoryProcessor],
-    //     });
+        const consumer = mongoDBEventStoreConsumer<ShoppingCartSummaryEvent>({
+          connectionString,
+          clientOptions: { directConnection: true },
+          processors: [inMemoryProcessor],
+        });
 
-    //     // When
-    //     await consumer.start();
-    //     await consumer.stop();
+        // When
+        await consumer.start();
+        await consumer.stop();
 
-    //     stopAfterPosition = undefined;
+        stopAfterPosition = undefined;
 
-    //     try {
-    //       const consumerPromise = consumer.start();
+        try {
+          const consumerPromise = consumer.start();
 
-    //       const appendResult = await eventStore.appendToStream(
-    //         streamName,
-    //         events,
-    //       );
-    //       stopAfterPosition = appendResult.nextExpectedStreamVersion;
+          const appendResult = await eventStore.appendToStream(
+            streamName,
+            events,
+          );
+          stopAfterPosition = appendResult.nextExpectedStreamVersion;
 
-    //       await consumerPromise;
+          await consumerPromise;
 
-    //       const summary = await summaries.findOne((d) => d._id === streamName);
+          const summary = await summaries.findOne((d) => d._id === streamName);
 
-    //       assertMatches(summary, {
-    //         _id: streamName,
-    //         status: 'confirmed',
-    //         //_version: 4n,
-    //         productItemsCount: productItem.quantity * 3,
-    //       });
-    //     } finally {
-    //       await consumer.close();
-    //     }
-    //   },
-    // );
+          assertMatches(summary, {
+            _id: streamName,
+            status: 'confirmed',
+            //_version: 4n,
+            productItemsCount: productItem.quantity * 3,
+          });
+        } finally {
+          await consumer.close();
+        }
+      },
+    );
 
     // void it(
     //   'handles only new events when CURRENT position is stored for a new consumer',

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventStoreConsumer.int.spec.ts

@@ -101,7 +101,10 @@ void describe('mongoDB event store consumer', () => {
         const connectionStringToNotExistingDB = 'mongodb://not-existing:2113';
         const consumerToNotExistingServer = mongoDBEventStoreConsumer({
           connectionString: connectionStringToNotExistingDB,
-          clientOptions: { directConnection: true },
+          clientOptions: {
+            directConnection: true,
+            serverSelectionTimeoutMS: 1000,
+          },
           processors: [dummyProcessor],
         });
         await assertThrowsAsync(

src/packages/emmett-mongodb/src/eventStore/consumers/mongoDBEventStoreConsumer.ts

@@ -14,7 +14,6 @@ import {
 } from '@event-driven-io/emmett';
 import { MongoClient, type MongoClientOptions } from 'mongodb';
 import { v4 as uuid } from 'uuid';
-import { CancellationPromise } from './CancellablePromise';
 import {
   changeStreamReactor,
   mongoDBProjector,
@@ -103,7 +102,6 @@ export const mongoDBEventStoreConsumer = <
   let start: Promise<void>;
   let stream: MongoDBSubscription | undefined;
   let isRunning = false;
-  let runningPromise = new CancellationPromise<null>();
   const client =
     'client' in options && options.client
       ? options.client
@@ -144,10 +142,11 @@ export const mongoDBEventStoreConsumer = <
   };
 
   const stop = async () => {
+    if (!isRunning) return;
+    isRunning = false;
+
     if (stream?.isRunning !== true) return;
     await stream.stop();
-    isRunning = false;
-    runningPromise.resolve(null);
   };
 
   return {
@@ -201,8 +200,6 @@ export const mongoDBEventStoreConsumer = <
 
         isRunning = true;
 
-        runningPromise = new CancellationPromise<null>();
-
         const positions = await Promise.all(
           processors.map((o) => o.start({ client })),
         );