Used SequentialTransform base class in ESDB consumer

Author: oskardudycz

src/packages/emmett-esdb/src/eventStore/consumers/subscriptions/index.ts

@@ -10,6 +10,11 @@ import {
   type Message,
   type MessageHandlerResult,
 } from '@event-driven-io/emmett';
+import {
+  SequentialTransform,
+  type SequentialTransformHandler,
+  type SequentialTransformHandlerResult,
+} from '@event-driven-io/emmett/nodejs';
 import {
   END,
   EventStoreDBClient,
@@ -18,7 +23,7 @@ import {
   type ResolvedEvent,
   type StreamSubscription,
 } from '@eventstore/db-client';
-import { pipeline, Transform, Writable, type WritableOptions } from 'stream';
+import { pipeline, Writable, type WritableOptions } from 'stream';
 import {
   mapFromESDBEvent,
   type EventStoreDBReadEventMetadata,
@@ -120,49 +125,45 @@ type SubscriptionSequentialHandlerOptions<
 
 class SubscriptionSequentialHandler<
   MessageType extends AnyMessage = AnyMessage,
-> extends Transform {
+> extends SequentialTransform<ResolvedEvent<MessageType>, bigint | null> {
   private options: SubscriptionSequentialHandlerOptions<MessageType>;
   private from: EventStoreDBEventStoreConsumerType | undefined;
   public isRunning: boolean;
 
   constructor(options: SubscriptionSequentialHandlerOptions<MessageType>) {
-    super({ objectMode: true, ...options });
-    this.options = options;
-    this.from = options.from;
-    this.isRunning = true;
-  }
-
-  async _transform(
-    resolvedEvent: ResolvedEvent<MessageType>,
-    _encoding: BufferEncoding,
-    callback: (error?: Error | null) => void,
-  ): Promise<void> {
-    try {
+    const handler: SequentialTransformHandler<
+      ResolvedEvent<MessageType>,
+      bigint
+    > = async (
+      resolvedEvent: ResolvedEvent<MessageType>,
+    ): Promise<SequentialTransformHandlerResult<bigint>> => {
       if (!this.isRunning || !resolvedEvent.event) {
-        callback();
-        return;
+        return { resultType: 'SKIP' };
       }
 
       const message = mapFromESDBEvent(resolvedEvent, this.from);
-      const messageCheckpoint = getCheckpoint(message);
+      const messageCheckpoint = getCheckpoint(message)!;
 
       const result = await this.options.eachBatch([message]);
 
       if (result && result.type === 'STOP') {
         this.isRunning = false;
-        if (!result.error) this.push(messageCheckpoint);
+        if (!result.error) {
+          return { resultType: 'ACK', message: messageCheckpoint };
+        }
 
-        this.push(result);
-        this.push(null);
-        callback();
-        return;
+        return { resultType: 'STOP', ...result };
       }
 
-      this.push(messageCheckpoint);
-      callback();
-    } catch (error) {
-      callback(error as Error);
-    }
+      return { resultType: 'ACK', message: messageCheckpoint };
+    };
+    super({
+      ...options,
+      handler,
+    });
+    this.options = options;
+    this.from = options.from;
+    this.isRunning = true;
   }
 }
 

src/packages/emmett/package.json

@@ -49,14 +49,14 @@
         "default": "./dist/cli.cjs"
       }
     },
-    "./fusionStreams": {
+    "./nodejs": {
       "import": {
-        "types": "./dist/fusionStreams.d.ts",
-        "default": "./dist/fusionStreams.js"
+        "types": "./dist/nodejs.d.ts",
+        "default": "./dist/nodejs.js"
       },
       "require": {
-        "types": "./dist/fusionStreams.d.cts",
-        "default": "./dist/fusionStreams.cjs"
+        "types": "./dist/nodejs.d.cts",
+        "default": "./dist/nodejs.cjs"
       }
     }
   },
@@ -68,8 +68,8 @@
       "cli": [
         "./dist/cli.d.ts"
       ],
-      "fusionStreams": [
-        "./dist/fusionStreams.d.ts"
+      "nodejs": [
+        "./dist/nodejs.d.ts"
       ]
     }
   },

src/packages/emmett/src/fusionStreams/transformations/index.ts

@@ -1 +1 @@
-export * from './sequentialTransformation';
+export * from './sequentialTransform';

…nsformations/sequentialTransformation.ts …s/transformations/sequentialTransform.tssrc/packages/emmett/src/fusionStreams/transformations/sequentialTransformation.ts renamed to src/packages/emmett/src/fusionStreams/transformations/sequentialTransform.ts src/packages/emmett/src/fusionStreams/transformations/sequentialTransformation.ts renamed to src/packages/emmett/src/fusionStreams/transformations/sequentialTransform.ts

@@ -1,13 +1,19 @@
 import { Transform, type WritableOptions } from 'stream';
+import type { EmmettError } from '../../errors';
+
+export type SequentialTransformHandlerResultType = 'ACK' | 'SKIP' | 'STOP';
+
+export type SequentialTransformHandlerResult<OutgoingMessageType = unknown> =
+  | { resultType: 'ACK'; message: OutgoingMessageType }
+  | { resultType: 'SKIP'; reason?: string }
+  | { resultType: 'STOP'; reason?: string; error?: EmmettError };
 
 export type SequentialTransformHandler<
   IncomingMessageType = unknown,
   OutgoingMessageType = unknown,
-> = {
-  handler: (
-    message: IncomingMessageType,
-  ) => Promise<OutgoingMessageType | null>;
-};
+> = (
+  message: IncomingMessageType,
+) => Promise<SequentialTransformHandlerResult<OutgoingMessageType>>;
 
 export type SequentialTransformOptions<
   IncomingMessageType = unknown,
@@ -40,9 +46,18 @@ export class SequentialTransform<
     callback: (error?: Error | null) => void,
   ): Promise<void> {
     try {
-      const result = await this.handler.handler(message);
+      const result = await this.handler(message);
 
-      this.push(result);
+      switch (result.resultType) {
+        case 'ACK':
+          this.push(result);
+          break;
+        case 'SKIP':
+          break;
+        case 'STOP':
+          this.push(null);
+          break;
+      }
 
       callback();
     } catch (error) {

src/packages/emmett/src/fusionStreams.ts src/packages/emmett/src/nodejs.tssrc/packages/emmett/src/fusionStreams.ts renamed to src/packages/emmett/src/nodejs.ts

@@ -15,7 +15,7 @@ export default defineConfig([
     watch: env === 'development',
     target: 'esnext',
     outDir: 'dist', //env === 'production' ? 'dist' : 'lib',
-    entry: ['src/index.ts', 'src/cli.ts', 'src/fusionStreams.ts'],
+    entry: ['src/index.ts', 'src/cli.ts', 'src/nodejs.ts'],
     sourcemap: true,
     tsconfig: 'tsconfig.build.json', // workaround for https://github.com/egoist/tsup/issues/571#issuecomment-1760052931
   },