Made consumer options really optional when creating it from event stores

By accident, there was a leftover forcing user to still provide even empty object, which doesn't make sense as it should be passed.
event-driven-io · Feb 14, 2025 · ef694ca · ef694ca
1 parent 909b36c
commit ef694ca
Show file tree

Hide file tree

Showing 5 changed files with 34 additions and 34 deletions.
diff --git a/src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.ts b/src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreConsumer.ts
@@ -20,7 +20,6 @@ import {
 export type EventStoreDBEventStoreConsumerConfig<
   ConsumerEventType extends Event = Event,
 > = {
-  connectionString: string;
   from?: EventStoreDBEventStoreConsumerType;
   processors?: EventStoreDBEventStoreProcessor<ConsumerEventType>[];
   pulling?: {
@@ -35,7 +34,7 @@ export type EventStoreDBEventStoreConsumerOptions<
     | {
         connectionString: string;
       }
-    | { eventStoreDBClient: EventStoreDBClient }
+    | { client: EventStoreDBClient }
   );
 
 export type $all = '$all';
@@ -54,7 +53,6 @@ export type EventStoreDBEventStoreConsumerType =
 export type EventStoreDBEventStoreConsumer<
   ConsumerEventType extends Event = Event,
 > = Readonly<{
-  connectionString: string;
   isRunning: boolean;
   processors: EventStoreDBEventStoreProcessor<ConsumerEventType>[];
   processor: <EventType extends ConsumerEventType = ConsumerEventType>(
@@ -71,17 +69,17 @@ export const eventStoreDBEventStoreConsumer = <
   options: EventStoreDBEventStoreConsumerOptions<ConsumerEventType>,
 ): EventStoreDBEventStoreConsumer<ConsumerEventType> => {
   let isRunning = false;
-  const { connectionString, pulling } = options;
+  const { pulling } = options;
   const processors = options.processors ?? [];
 
   let start: Promise<void>;
 
-  let currentMessagePooler:
-    | EventStoreDBEventStoreMessageBatchPuller
-    | undefined;
+  let currentSubscription: EventStoreDBEventStoreMessageBatchPuller | undefined;
 
-  const eventStoreDBClient =
-    EventStoreDBClient.connectionString(connectionString);
+  const client =
+    'client' in options
+      ? options.client
+      : EventStoreDBClient.connectionString(options.connectionString);
 
   const eachBatch: EventStoreDBEventStoreMessagesBatchHandler<
     ConsumerEventType
@@ -97,7 +95,7 @@ export const eventStoreDBEventStoreConsumer = <
     const result = await Promise.allSettled(
       activeProcessors.map((s) => {
         // TODO: Add here filtering to only pass messages that can be handled by processor
-        return s.handle(messagesBatch, { eventStoreDBClient });
+        return s.handle(messagesBatch, { client });
       }),
     );
 
@@ -110,8 +108,8 @@ export const eventStoreDBEventStoreConsumer = <
         };
   };
 
-  const messagePuller = (currentMessagePooler = eventStoreDBSubscription({
-    eventStoreDBClient,
+  const subscription = (currentSubscription = eventStoreDBSubscription({
+    client,
     eachBatch,
     batchSize:
       pulling?.batchSize ?? DefaultEventStoreDBEventStoreProcessorBatchSize,
@@ -120,15 +118,14 @@ export const eventStoreDBEventStoreConsumer = <
   const stop = async () => {
     if (!isRunning) return;
     isRunning = false;
-    if (currentMessagePooler) {
-      await currentMessagePooler.stop();
-      currentMessagePooler = undefined;
+    if (currentSubscription) {
+      await currentSubscription.stop();
+      currentSubscription = undefined;
     }
     await start;
   };
 
   return {
-    connectionString,
     processors,
     get isRunning() {
       return isRunning;
@@ -156,10 +153,10 @@ export const eventStoreDBEventStoreConsumer = <
         isRunning = true;
 
         const startFrom = zipEventStoreDBEventStoreMessageBatchPullerStartFrom(
-          await Promise.all(processors.map((o) => o.start(eventStoreDBClient))),
+          await Promise.all(processors.map((o) => o.start(client))),
         );
 
-        return messagePuller.start({ startFrom });
+        return subscription.start({ startFrom });
       })();
 
       return start;

diff --git a/src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreProcessor.ts b/src/packages/emmett-esdb/src/eventStore/consumers/eventStoreDBEventStoreProcessor.ts
@@ -17,12 +17,12 @@ export type EventStoreDBEventStoreProcessorEventsBatch<
 export type EventStoreDBEventStoreProcessor<EventType extends Event = Event> = {
   id: string;
   start: (
-    eventStoreDBClient: EventStoreDBClient,
+    client: EventStoreDBClient,
   ) => Promise<EventStoreDBSubscriptionStartFrom | undefined>;
   isActive: boolean;
   handle: (
     messagesBatch: EventStoreDBEventStoreProcessorEventsBatch<EventType>,
-    context: { eventStoreDBClient: EventStoreDBClient },
+    context: { client: EventStoreDBClient },
   ) => Promise<EventStoreDBEventStoreProcessorMessageHandlerResult>;
 };
 
@@ -88,7 +88,7 @@ export const eventStoreDBEventStoreProcessor = <
   return {
     id: options.processorId,
     start: (
-      _eventStoreDBClient: EventStoreDBClient,
+      _client: EventStoreDBClient,
     ): Promise<EventStoreDBSubscriptionStartFrom | undefined> => {
       isActive = true;
       if (options.startFrom !== 'CURRENT')

diff --git a/src/packages/emmett-esdb/src/eventStore/consumers/subscriptions/index.ts b/src/packages/emmett-esdb/src/eventStore/consumers/subscriptions/index.ts
@@ -44,7 +44,7 @@ export type EventStoreDBEventStoreMessagesBatchHandler<
 
 export type EventStoreDBSubscriptionOptions<EventType extends Event = Event> = {
   from?: EventStoreDBEventStoreConsumerType;
-  eventStoreDBClient: EventStoreDBClient;
+  client: EventStoreDBClient;
   batchSize: number;
   eachBatch: EventStoreDBEventStoreMessagesBatchHandler<EventType>;
 };
@@ -82,23 +82,23 @@ const toStreamPosition = (startFrom: EventStoreDBSubscriptionStartFrom) =>
       : startFrom.position;
 
 const subscribe = (
-  eventStoreDBClient: EventStoreDBClient,
+  client: EventStoreDBClient,
   from: EventStoreDBEventStoreConsumerType | undefined,
   options: EventStoreDBSubscriptionStartOptions,
 ) =>
   from == undefined || from.stream == $all
-    ? eventStoreDBClient.subscribeToAll({
+    ? client.subscribeToAll({
         fromPosition: toGlobalPosition(options.startFrom),
         filter: excludeSystemEvents(),
         ...(from?.options ?? {}),
       })
-    : eventStoreDBClient.subscribeToStream(from.stream, {
+    : client.subscribeToStream(from.stream, {
         fromRevision: toStreamPosition(options.startFrom),
         ...(from.options ?? {}),
       });
 
 export const eventStoreDBSubscription = <EventType extends Event = Event>({
-  eventStoreDBClient,
+  client,
   from,
   //batchSize,
   eachBatch,
@@ -112,7 +112,7 @@ export const eventStoreDBSubscription = <EventType extends Event = Event>({
   const pullMessages = async (
     options: EventStoreDBSubscriptionStartOptions,
   ) => {
-    subscription = subscribe(eventStoreDBClient, from, options);
+    subscription = subscribe(client, from, options);
 
     return new Promise<void>((resolve, reject) => {
       finished(

diff --git a/src/packages/emmett-esdb/src/eventStore/eventstoreDBEventStore.ts b/src/packages/emmett-esdb/src/eventStore/eventstoreDBEventStore.ts
@@ -67,7 +67,7 @@ export interface EventStoreDBEventStore
     options?: AppendToStreamOptions,
   ): Promise<AppendToStreamResultWithGlobalPosition>;
   consumer<ConsumerEventType extends Event = Event>(
-    options: EventStoreDBEventStoreConsumerConfig<ConsumerEventType>,
+    options?: EventStoreDBEventStoreConsumerConfig<ConsumerEventType>,
   ): EventStoreDBEventStoreConsumer<ConsumerEventType>;
 }
 
@@ -211,11 +211,11 @@ export const getEventStoreDBEventStore = (
     },
 
     consumer: <ConsumerEventType extends Event = Event>(
-      options: EventStoreDBEventStoreConsumerConfig<ConsumerEventType>,
+      options?: EventStoreDBEventStoreConsumerConfig<ConsumerEventType>,
     ): EventStoreDBEventStoreConsumer<ConsumerEventType> =>
       eventStoreDBEventStoreConsumer<ConsumerEventType>({
-        ...options,
-        eventStoreDBClient: eventStore,
+        ...(options ?? {}),
+        client: eventStore,
       }),
 
     //streamEvents: streamEvents(eventStore),

diff --git a/src/packages/emmett-postgresql/src/eventStore/postgreSQLEventStore.ts b/src/packages/emmett-postgresql/src/eventStore/postgreSQLEventStore.ts
@@ -51,7 +51,7 @@ export interface PostgresEventStore
     options?: AppendToStreamOptions,
   ): Promise<AppendToStreamResultWithGlobalPosition>;
   consumer<ConsumerEventType extends Event = Event>(
-    options: PostgreSQLEventStoreConsumerConfig<ConsumerEventType>,
+    options?: PostgreSQLEventStoreConsumerConfig<ConsumerEventType>,
   ): PostgreSQLEventStoreConsumer<ConsumerEventType>;
   close(): Promise<void>;
   schema: {
@@ -284,9 +284,12 @@ export const getPostgreSQLEventStore = (
       };
     },
     consumer: <ConsumerEventType extends Event = Event>(
-      options: PostgreSQLEventStoreConsumerConfig<ConsumerEventType>,
+      options?: PostgreSQLEventStoreConsumerConfig<ConsumerEventType>,
     ): PostgreSQLEventStoreConsumer<ConsumerEventType> =>
-      postgreSQLEventStoreConsumer<ConsumerEventType>({ ...options, pool }),
+      postgreSQLEventStoreConsumer<ConsumerEventType>({
+        ...(options ?? {}),
+        pool,
+      }),
     close: () => pool.close(),
 
     async withSession<T = unknown>(