Added task processing

Author: oskardudycz

src/packages/emmett-postgresql/src/eventStore/projections/postgresProjection.multi.int.spec.ts

@@ -1,4 +1,4 @@
-import type { ReadEvent } from '@event-driven-io/emmett/src';
+import type { ReadEvent } from '@event-driven-io/emmett';
 import {
   PostgreSqlContainer,
   StartedPostgreSqlContainer,

src/packages/emmett-testcontainers/src/eventStore/eventStoreDBContainer.ts

@@ -1,3 +1,4 @@
+import { InProcessLock } from '@event-driven-io/emmett';
 import { EventStoreDBClient } from '@eventstore/db-client';
 import {
   AbstractStartedContainer,
@@ -9,8 +10,8 @@ import type { Environment } from 'testcontainers/build/types';
 
 export const EVENTSTOREDB_PORT = 2113;
 export const EVENTSTOREDB_IMAGE_NAME = 'eventstore/eventstore';
-export const EVENTSTOREDB_IMAGE_TAG = '23.10.1-bookworm-slim';
-export const EVENTSTOREDB_ARM64_IMAGE_TAG = '23.10.1-alpha-arm64v8';
+export const EVENTSTOREDB_IMAGE_TAG = '24.10.0-bookworm-slim';
+export const EVENTSTOREDB_ARM64_IMAGE_TAG = '24.10.0-alpha-arm64v8';
 
 export const EVENTSTOREDB_DEFAULT_IMAGE = `${EVENTSTOREDB_IMAGE_NAME}:${process.arch !== 'arm64' ? EVENTSTOREDB_IMAGE_TAG : EVENTSTOREDB_ARM64_IMAGE_TAG}`;
 
@@ -86,44 +87,52 @@ export class StartedEventStoreDBContainer extends AbstractStartedContainer {
     return EventStoreDBClient.connectionString(this.getConnectionString());
   }
 }
+
 let container: EventStoreDBContainer | null = null;
 let startedContainer: StartedEventStoreDBContainer | null = null;
 let startedCount = 0;
+const lock = InProcessLock();
 
-export const getSharedEventStoreDBTestContainer = async () => {
-  if (startedContainer) return startedContainer;
+export const getSharedEventStoreDBTestContainer = () =>
+  lock.withAcquire(
+    async () => {
+      if (startedContainer) return startedContainer;
 
-  if (!container)
-    container = new EventStoreDBContainer(EVENTSTOREDB_DEFAULT_IMAGE, {
-      withReuse: true,
-    });
+      if (!container)
+        container = new EventStoreDBContainer(EVENTSTOREDB_DEFAULT_IMAGE);
 
-  startedContainer = await container.start();
-  startedCount++;
+      startedContainer = await container.start();
+      startedCount++;
 
-  container.withLogConsumer((stream) =>
-    stream
-      .on('data', (line) => console.log(line))
-      .on('err', (line) => console.error(line))
-      .on('end', () => console.log('Stream closed')),
-  );
+      container.withLogConsumer((stream) =>
+        stream
+          .on('data', (line) => console.log(line))
+          .on('err', (line) => console.error(line))
+          .on('end', () => console.log('Stream closed')),
+      );
 
-  return startedContainer;
-};
+      return startedContainer;
+    },
+    { lockId: 'SharedEventStoreDBTestContainer' },
+  );
 
 export const getSharedTestEventStoreDBClient = async () => {
   return (await getSharedEventStoreDBTestContainer()).getClient();
 };
 
-export const releaseSharedEventStoreDBTestContainer = async () => {
-  const containerToStop = startedContainer;
-  if (containerToStop && --startedCount === 0) {
-    try {
-      startedContainer = null;
-      container = null;
-      await containerToStop.stop();
-    } catch {
-      /* do nothing */
-    }
-  }
-};
+export const releaseSharedEventStoreDBTestContainer = () =>
+  lock.withAcquire(
+    async () => {
+      const containerToStop = startedContainer;
+      if (containerToStop && --startedCount === 0) {
+        try {
+          startedContainer = null;
+          container = null;
+          await containerToStop.stop();
+        } catch {
+          /* do nothing */
+        }
+      }
+    },
+    { lockId: 'SharedEventStoreDBTestContainer' },
+  );

src/packages/emmett-tests/src/eventStore/esdb/eventstoreDBEventStore.e2e.spec.ts

@@ -1,7 +1,8 @@
 //import { streamTransformations, type Event } from '@event-driven-io/emmett';
 import { getEventStoreDBEventStore } from '@event-driven-io/emmett-esdb';
 import {
-  EventStoreDBContainer,
+  getSharedEventStoreDBTestContainer,
+  releaseSharedEventStoreDBTestContainer,
   StartedEventStoreDBContainer,
 } from '@event-driven-io/emmett-testcontainers';
 import { describe } from 'node:test';
@@ -15,12 +16,12 @@ void describe('EventStoreDBEventStore', async () => {
   let esdbContainer: StartedEventStoreDBContainer;
 
   const eventStoreFactory: EventStoreFactory = async () => {
-    esdbContainer = await new EventStoreDBContainer().start();
+    esdbContainer = await getSharedEventStoreDBTestContainer();
     return getEventStoreDBEventStore(esdbContainer.getClient());
   };
 
   const teardownHook = async () => {
-    await esdbContainer.stop();
+    await releaseSharedEventStoreDBTestContainer();
   };
 
   await testAggregateStream(eventStoreFactory, {

src/packages/emmett/src/index.ts

@@ -7,6 +7,7 @@ export * from './projections';
 export * from './serialization';
 export * from './streaming';
 export * from './subscriptions';
+export * from './taskProcessing';
 export * from './testing';
 export * from './typing';
 export * from './utils';

src/packages/emmett/src/taskProcessing/index.ts

@@ -0,0 +1 @@
+export * from './taskProcessor';

src/packages/emmett/src/taskProcessing/taskProcessor.ts

@@ -0,0 +1,186 @@
+import { EmmettError } from '../errors';
+
+export type TaskQueue = TaskQueueItem[];
+
+export type TaskQueueItem = {
+  task: () => Promise<void>;
+  options?: EnqueueTaskOptions;
+};
+
+export type TaskProcessorOptions = {
+  maxActiveTasks: number;
+  maxQueueSize: number;
+  maxTaskIdleTime?: number;
+};
+
+export type Task<T> = (context: TaskContext) => Promise<T>;
+
+export type TaskContext = {
+  ack: () => void;
+};
+
+export type EnqueueTaskOptions = { taskGroupId?: string };
+
+export class TaskProcessor {
+  private queue: TaskQueue = [];
+  private isProcessing = false;
+  private activeTasks = 0;
+  private activeGroups: Set<string> = new Set();
+
+  constructor(private options: TaskProcessorOptions) {}
+
+  enqueue<T>(task: Task<T>, options?: EnqueueTaskOptions): Promise<T> {
+    if (this.queue.length >= this.options.maxQueueSize) {
+      return Promise.reject(
+        new EmmettError(
+          'Too many pending connections. Please try again later.',
+        ),
+      );
+    }
+
+    return this.schedule(task, options);
+  }
+
+  waitForEndOfProcessing(): Promise<void> {
+    return this.schedule(({ ack }) => Promise.resolve(ack()));
+  }
+
+  private schedule<T>(task: Task<T>, options?: EnqueueTaskOptions): Promise<T> {
+    return promiseWithDeadline(
+      (resolve, reject) => {
+        const taskWithContext = () => {
+          return new Promise<void>((resolveTask, failTask) => {
+            const taskPromise = task({
+              ack: resolveTask,
+            });
+
+            taskPromise.then(resolve).catch((err) => {
+              // eslint-disable-next-line @typescript-eslint/prefer-promise-reject-errors
+              failTask(err);
+              reject(err);
+            });
+          });
+        };
+
+        this.queue.push({ task: taskWithContext, options });
+        if (!this.isProcessing) {
+          this.ensureProcessing();
+        }
+      },
+      { deadline: this.options.maxTaskIdleTime },
+    );
+  }
+
+  private ensureProcessing(): void {
+    if (this.isProcessing) return;
+    this.isProcessing = true;
+    this.processQueue();
+  }
+
+  private processQueue(): void {
+    try {
+      while (
+        this.activeTasks < this.options.maxActiveTasks &&
+        this.queue.length > 0
+      ) {
+        const item = this.takeFirstAvailableItem();
+
+        if (item === null) return;
+
+        const groupId = item.options?.taskGroupId;
+
+        if (groupId) {
+          // Mark the group as active
+          this.activeGroups.add(groupId);
+        }
+
+        this.activeTasks++;
+        void this.executeItem(item);
+      }
+    } catch (error) {
+      console.error(error);
+      throw error;
+    } finally {
+      this.isProcessing = false;
+      if (
+        this.hasItemsToProcess() &&
+        this.activeTasks < this.options.maxActiveTasks
+      ) {
+        this.ensureProcessing();
+      }
+    }
+  }
+
+  private async executeItem({ task, options }: TaskQueueItem): Promise<void> {
+    try {
+      await task();
+    } finally {
+      this.activeTasks--;
+
+      // Mark the group as inactive after task completion
+      if (options && options.taskGroupId) {
+        this.activeGroups.delete(options.taskGroupId);
+      }
+
+      this.ensureProcessing();
+    }
+  }
+
+  private takeFirstAvailableItem = (): TaskQueueItem | null => {
+    const taskIndex = this.queue.findIndex(
+      (item) =>
+        !item.options?.taskGroupId ||
+        !this.activeGroups.has(item.options.taskGroupId),
+    );
+
+    if (taskIndex === -1) {
+      // All remaining tasks are blocked by active groups
+      return null;
+    }
+
+    // Remove the task from the queue
+    const [item] = this.queue.splice(taskIndex, 1);
+
+    return item ?? null;
+  };
+
+  private hasItemsToProcess = (): boolean =>
+    this.queue.findIndex(
+      (item) =>
+        !item.options?.taskGroupId ||
+        !this.activeGroups.has(item.options.taskGroupId),
+    ) !== -1;
+}
+
+const DEFAULT_PROMISE_DEADLINE = 2147483647;
+
+const promiseWithDeadline = <T>(
+  executor: (
+    resolve: (value: T | PromiseLike<T>) => void,
+    reject: (reason?: unknown) => void,
+  ) => void,
+  options: { deadline?: number },
+) => {
+  return new Promise<T>((resolve, reject) => {
+    let taskStarted = false;
+
+    const maxWaitingTime = options.deadline || DEFAULT_PROMISE_DEADLINE;
+
+    let timeoutId: NodeJS.Timeout | null = setTimeout(() => {
+      if (!taskStarted) {
+        reject(
+          new Error('Task was not started within the maximum waiting time'),
+        );
+      }
+    }, maxWaitingTime);
+
+    executor((value) => {
+      taskStarted = true;
+      if (timeoutId) {
+        clearTimeout(timeoutId);
+      }
+      timeoutId = null;
+      resolve(value);
+    }, reject);
+  });
+};