@thecolvinco/nodejs-messenger - npm Package Compare versions

Comparing version 1.2.0 to 1.4.0

dist/consumer/CommandConsumer.d.ts

 /// <reference types="node" />
-import { config as configType } from '../types';
+import { config as configType, messageBody } from '../types';
 import { EventEmitter } from 'events';
@@ -10,3 +10,3 @@ import { Channel, Message } from 'amqplib';
     });
-    consume({ transport, queueName, prefetchValue, emitter, onError, }: {
+    consume({ transport, queueName, prefetchValue, emitter, onError, eventualConsistency, }: {
         transport: string;
@@ -16,5 +16,12 @@ queueName: string;
         emitter?: EventEmitter;
-        onError: (error: Error) => void;
+        onError: ({ error, message }: {
+            error: Error;
+            message?: messageBody;
+        }) => Promise<void>;
+        eventualConsistency?: {
+            isConsistent: (message: messageBody) => Promise<boolean>;
+            saveMessage?: (message: messageBody) => Promise<void>;
+        };
     }): Promise<void>;
-    workable({ channel, emitter, retryPolicy, queueName, onError }: {
+    workable({ channel, emitter, retryPolicy, queueName, onError, eventualConsistency, }: {
         channel: Channel;
@@ -29,4 +36,11 @@ emitter: EventEmitter;
         queueName: string;
-        onError: (error: Error) => void;
+        onError: ({ error, message }: {
+            error: Error;
+            message: messageBody;
+        }) => Promise<void>;
+        eventualConsistency: {
+            isConsistent: (message: messageBody) => Promise<boolean>;
+            saveMessage?: (message: messageBody) => Promise<void>;
+        };
     }): (msg: Message) => Promise<void>;
 }

dist/consumer/DomainEventConsumer.d.ts

 /// <reference types="node" />
-import { config as configType } from '../types';
+import { config as configType, messageBody } from '../types';
 import { EventEmitter } from 'events';
@@ -10,3 +10,3 @@ import { Channel, Message } from 'amqplib';
     });
-    consume({ transport, queueName, prefetchValue, emitter, onError, }: {
+    consume({ transport, queueName, prefetchValue, emitter, onError, eventualConsistency, }: {
         transport: string;
@@ -16,9 +16,23 @@ queueName: string;
         emitter?: EventEmitter;
-        onError: (error: Error) => void;
+        onError: ({ error, message }: {
+            error: Error;
+            message?: messageBody;
+        }) => Promise<void>;
+        eventualConsistency?: {
+            isConsistent: (message: messageBody) => Promise<boolean>;
+            saveMessage?: (message: messageBody) => Promise<void>;
+        };
     }): Promise<void>;
-    workable({ channel, emitter, onError }: {
+    workable({ channel, emitter, onError, eventualConsistency }: {
         channel: Channel;
         emitter: EventEmitter;
-        onError: (error: Error) => void;
+        onError: ({ error, message }: {
+            error: Error;
+            message: messageBody;
+        }) => Promise<void>;
+        eventualConsistency: {
+            isConsistent: (message: messageBody) => Promise<boolean>;
+            saveMessage?: (message: messageBody) => Promise<void>;
+        };
     }): (msg: Message) => Promise<void>;
 }

dist/Events/Event.d.ts

@@ -11,2 +11,3 @@ import { EventInterface } from '../Interfaces';
     abstract getPayload(): unknown;
+    abstract getEntityId(): string;
     protected getVersion(): string;
@@ -13,0 +14,0 @@ getName(): string;

dist/main.d.ts

@@ -8,2 +8,3 @@ import { createEvent, createCommand, toJSON, producer, worker, deadLetter, retryable, amqpConnect, MessageDeleter, MessageShifter } from './utils';
 import { EventHandlerInterface, EventInterface } from './Interfaces';
-export { amqpConnect, retryable, deadLetter, worker, producer, createEvent, createCommand, toJSON, MessageDeleter, MessageShifter, CommandConsumer, DomainEventConsumer, pubSubInitialization, transportsInitialization, CommandEmitter, DomainEventEmitter, EventDispatcher, commandSubscriber, domainEventSubscriber, Event, Command, DomainEvent, EventHandlerInterface, EventInterface, };
+import { messageBody } from './types';
+export { amqpConnect, retryable, deadLetter, worker, producer, createEvent, createCommand, toJSON, MessageDeleter, MessageShifter, CommandConsumer, DomainEventConsumer, pubSubInitialization, transportsInitialization, CommandEmitter, DomainEventEmitter, EventDispatcher, commandSubscriber, domainEventSubscriber, Event, Command, DomainEvent, EventHandlerInterface, EventInterface, messageBody, };

dist/main.esm.js

@@ -9,3 +9,4 @@ import { v4 } from 'uuid';
   meta,
-  type
+  type,
+  entityId
 }) => {
@@ -23,2 +24,3 @@ const {
       occurredOn: Date.now(),
+      entityId,
       attributes: payload,
@@ -32,3 +34,4 @@ type: `${company}.${context}.${version}.${type}.${entity}.${name}`
   payload,
-  meta
+  meta,
+  entityId
 }) => {
@@ -38,3 +41,4 @@ return parse({
     meta,
-    type: 'domain_event'
+    type: 'domain_event',
+    entityId
   });
@@ -45,3 +49,4 @@ };
   payload,
-  meta
+  meta,
+  entityId
 }) => {
@@ -51,3 +56,4 @@ return parse({
     meta,
-    type: 'command'
+    type: 'command',
+    entityId
   });
@@ -509,3 +515,4 @@ };
           name
-        }
+        },
+        entityId: event.getEntityId()
       });
@@ -531,3 +538,4 @@ return this.asyncDispatch({
           name
-        }
+        },
+        entityId: event.getEntityId()
       });
@@ -586,3 +594,3 @@ return this.asyncDispatch({
       } catch (error) {
-        if (onError) onError({
+        if (onError) await onError({
           error
@@ -610,3 +618,3 @@ });
       } catch (error) {
-        if (onError) onError({
+        if (onError) await onError({
           error
@@ -767,3 +775,7 @@ });
     emitter = null,
-    onError
+    onError,
+    eventualConsistency = {
+      isConsistent: async message => true,
+      saveMessage: async message => undefined
+    }
   }) {
@@ -804,3 +816,4 @@ if (!this.transports[transport]) {
         queueName,
-        onError
+        onError,
+        eventualConsistency
       });
@@ -811,3 +824,5 @@ channel.consume(queueName, onMessage, {
     } catch (error) {
-      if (onError) onError(error);
+      if (onError) await onError({
+        error
+      });
       connection.close();
@@ -823,3 +838,4 @@ throw error;
     queueName,
-    onError
+    onError,
+    eventualConsistency
   }) {
@@ -836,7 +852,10 @@ return async msg => {
       try {
-        const onErrorCallback = ({
+        const onErrorCallback = async ({
           error
         }) => {
           if (!retryPolicy) channel.nack(msg, false, false);
-          if (onError) onError(error);
+          if (onError) await onError({
+            error,
+            message
+          });
           const {
@@ -867,2 +886,10 @@ maxRetries,
 
+        const isConsistent = await eventualConsistency.isConsistent(message);
+
+        if (!isConsistent) {
+          channel.ack(msg);
+          return;
+        }
+
+        eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
         emitter.emit(eventName, {
@@ -874,3 +901,6 @@ message,
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error,
+          message
+        });
         channel.nack(msg, false, false);
@@ -895,3 +925,7 @@ }
     emitter = null,
-    onError
+    onError,
+    eventualConsistency = {
+      isConsistent: async message => true,
+      saveMessage: async message => undefined
+    }
   }) {
@@ -928,3 +962,4 @@ if (!this.transports[transport]) {
           emitter: domainEventsEmitter,
-          onError
+          onError,
+          eventualConsistency
         }),
@@ -937,3 +972,5 @@ options: {
     } catch (error) {
-      if (onError) onError(error);
+      if (onError) await onError({
+        error
+      });
       connection.close();
@@ -947,3 +984,4 @@ throw error;
     emitter,
-    onError
+    onError,
+    eventualConsistency
   }) {
@@ -960,8 +998,19 @@ return async msg => {
       try {
+        const isConsistent = await eventualConsistency.isConsistent(message);
+
+        if (!isConsistent) {
+          channel.ack(msg);
+          return;
+        }
+
         emitter.emit(eventName, {
           message
         });
+        eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
         channel.ack(msg);
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error,
+          message
+        });
         console.error(`Domain event consumer error occurred: ${error.message}`);
@@ -968,0 +1017,0 @@ channel.nack(msg, false, false);

dist/main.js

@@ -28,3 +28,4 @@ function _interopNamespace(e) {
   meta,
-  type
+  type,
+  entityId
 }) => {
@@ -42,2 +43,3 @@ const {
       occurredOn: Date.now(),
+      entityId,
       attributes: payload,
@@ -51,3 +53,4 @@ type: `${company}.${context}.${version}.${type}.${entity}.${name}`
   payload,
-  meta
+  meta,
+  entityId
 }) => {
@@ -57,3 +60,4 @@ return parse({
     meta,
-    type: 'domain_event'
+    type: 'domain_event',
+    entityId
   });
@@ -64,3 +68,4 @@ };
   payload,
-  meta
+  meta,
+  entityId
 }) => {
@@ -70,3 +75,4 @@ return parse({
     meta,
-    type: 'command'
+    type: 'command',
+    entityId
   });
@@ -528,3 +534,4 @@ };
           name
-        }
+        },
+        entityId: event.getEntityId()
       });
@@ -550,3 +557,4 @@ return this.asyncDispatch({
           name
-        }
+        },
+        entityId: event.getEntityId()
       });
@@ -605,3 +613,3 @@ return this.asyncDispatch({
       } catch (error) {
-        if (onError) onError({
+        if (onError) await onError({
           error
@@ -629,3 +637,3 @@ });
       } catch (error) {
-        if (onError) onError({
+        if (onError) await onError({
           error
@@ -786,3 +794,7 @@ });
     emitter = null,
-    onError
+    onError,
+    eventualConsistency = {
+      isConsistent: async message => true,
+      saveMessage: async message => undefined
+    }
   }) {
@@ -823,3 +835,4 @@ if (!this.transports[transport]) {
         queueName,
-        onError
+        onError,
+        eventualConsistency
       });
@@ -830,3 +843,5 @@ channel.consume(queueName, onMessage, {
     } catch (error) {
-      if (onError) onError(error);
+      if (onError) await onError({
+        error
+      });
       connection.close();
@@ -842,3 +857,4 @@ throw error;
     queueName,
-    onError
+    onError,
+    eventualConsistency
   }) {
@@ -855,7 +871,10 @@ return async msg => {
       try {
-        const onErrorCallback = ({
+        const onErrorCallback = async ({
           error
         }) => {
           if (!retryPolicy) channel.nack(msg, false, false);
-          if (onError) onError(error);
+          if (onError) await onError({
+            error,
+            message
+          });
           const {
@@ -886,2 +905,10 @@ maxRetries,
 
+        const isConsistent = await eventualConsistency.isConsistent(message);
+
+        if (!isConsistent) {
+          channel.ack(msg);
+          return;
+        }
+
+        eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
         emitter.emit(eventName, {
@@ -893,3 +920,6 @@ message,
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error,
+          message
+        });
         channel.nack(msg, false, false);
@@ -914,3 +944,7 @@ }
     emitter = null,
-    onError
+    onError,
+    eventualConsistency = {
+      isConsistent: async message => true,
+      saveMessage: async message => undefined
+    }
   }) {
@@ -947,3 +981,4 @@ if (!this.transports[transport]) {
           emitter: domainEventsEmitter,
-          onError
+          onError,
+          eventualConsistency
         }),
@@ -956,3 +991,5 @@ options: {
     } catch (error) {
-      if (onError) onError(error);
+      if (onError) await onError({
+        error
+      });
       connection.close();
@@ -966,3 +1003,4 @@ throw error;
     emitter,
-    onError
+    onError,
+    eventualConsistency
   }) {
@@ -979,8 +1017,19 @@ return async msg => {
       try {
+        const isConsistent = await eventualConsistency.isConsistent(message);
+
+        if (!isConsistent) {
+          channel.ack(msg);
+          return;
+        }
+
         emitter.emit(eventName, {
           message
         });
+        eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
         channel.ack(msg);
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error,
+          message
+        });
         console.error(`Domain event consumer error occurred: ${error.message}`);
@@ -987,0 +1036,0 @@ channel.nack(msg, false, false);

dist/main.modern.js

@@ -9,3 +9,4 @@ import { v4 } from 'uuid';
   meta,
-  type
+  type,
+  entityId
 }) => {
@@ -23,2 +24,3 @@ const {
       occurredOn: Date.now(),
+      entityId,
       attributes: payload,
@@ -32,3 +34,4 @@ type: `${company}.${context}.${version}.${type}.${entity}.${name}`
   payload,
-  meta
+  meta,
+  entityId
 }) => {
@@ -38,3 +41,4 @@ return parse({
     meta,
-    type: 'domain_event'
+    type: 'domain_event',
+    entityId
   });
@@ -45,3 +49,4 @@ };
   payload,
-  meta
+  meta,
+  entityId
 }) => {
@@ -51,3 +56,4 @@ return parse({
     meta,
-    type: 'command'
+    type: 'command',
+    entityId
   });
@@ -511,3 +517,4 @@ };
           name
-        }
+        },
+        entityId: event.getEntityId()
       });
@@ -533,3 +540,4 @@ return _this.asyncDispatch({
           name
-        }
+        },
+        entityId: event.getEntityId()
       });
@@ -588,3 +596,3 @@ return _this.asyncDispatch({
       } catch (error) {
-        if (onError) onError({
+        if (onError) await onError({
           error
@@ -612,3 +620,3 @@ });
       } catch (error) {
-        if (onError) onError({
+        if (onError) await onError({
           error
@@ -769,3 +777,11 @@ });
     emitter = null,
-    onError
+    onError,
+    eventualConsistency = {
+      isConsistent: async function (message) {
+        return true;
+      },
+      saveMessage: async function (message) {
+        return undefined;
+      }
+    }
   }) {
@@ -806,3 +822,4 @@ if (!this.transports[transport]) {
         queueName,
-        onError
+        onError,
+        eventualConsistency
       });
@@ -813,3 +830,5 @@ channel.consume(queueName, onMessage, {
     } catch (error) {
-      if (onError) onError(error);
+      if (onError) await onError({
+        error
+      });
       connection.close();
@@ -825,3 +844,4 @@ throw error;
     queueName,
-    onError
+    onError,
+    eventualConsistency
   }) {
@@ -838,7 +858,10 @@ return async function (msg) {
       try {
-        const onErrorCallback = ({
+        const onErrorCallback = async function onErrorCallback({
           error
-        }) => {
+        }) {
           if (!retryPolicy) channel.nack(msg, false, false);
-          if (onError) onError(error);
+          if (onError) await onError({
+            error,
+            message
+          });
           const {
@@ -869,2 +892,10 @@ maxRetries,
 
+        const isConsistent = await eventualConsistency.isConsistent(message);
+
+        if (!isConsistent) {
+          channel.ack(msg);
+          return;
+        }
+
+        eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
         emitter.emit(eventName, {
@@ -876,3 +907,6 @@ message,
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error,
+          message
+        });
         channel.nack(msg, false, false);
@@ -897,3 +931,11 @@ }
     emitter = null,
-    onError
+    onError,
+    eventualConsistency = {
+      isConsistent: async function (message) {
+        return true;
+      },
+      saveMessage: async function (message) {
+        return undefined;
+      }
+    }
   }) {
@@ -930,3 +972,4 @@ if (!this.transports[transport]) {
           emitter: domainEventsEmitter,
-          onError
+          onError,
+          eventualConsistency
         }),
@@ -939,3 +982,5 @@ options: {
     } catch (error) {
-      if (onError) onError(error);
+      if (onError) await onError({
+        error
+      });
       connection.close();
@@ -949,3 +994,4 @@ throw error;
     emitter,
-    onError
+    onError,
+    eventualConsistency
   }) {
@@ -962,8 +1008,19 @@ return async function (msg) {
       try {
+        const isConsistent = await eventualConsistency.isConsistent(message);
+
+        if (!isConsistent) {
+          channel.ack(msg);
+          return;
+        }
+
         emitter.emit(eventName, {
           message
         });
+        eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
         channel.ack(msg);
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error,
+          message
+        });
         console.error(`Domain event consumer error occurred: ${error.message}`);
@@ -970,0 +1027,0 @@ channel.nack(msg, false, false);

dist/main.umd.js

@@ -9,3 +9,4 @@ (function (global, factory) {
     meta,
-    type
+    type,
+    entityId
   }) => {
@@ -23,2 +24,3 @@ const {
         occurredOn: Date.now(),
+        entityId,
         attributes: payload,
@@ -32,3 +34,4 @@ type: `${company}.${context}.${version}.${type}.${entity}.${name}`
     payload,
-    meta
+    meta,
+    entityId
   }) => {
@@ -38,3 +41,4 @@ return parse({
       meta,
-      type: 'domain_event'
+      type: 'domain_event',
+      entityId
     });
@@ -45,3 +49,4 @@ };
     payload,
-    meta
+    meta,
+    entityId
   }) => {
@@ -51,3 +56,4 @@ return parse({
       meta,
-      type: 'command'
+      type: 'command',
+      entityId
     });
@@ -509,3 +515,4 @@ };
             name
-          }
+          },
+          entityId: event.getEntityId()
         });
@@ -531,3 +538,4 @@ return this.asyncDispatch({
             name
-          }
+          },
+          entityId: event.getEntityId()
         });
@@ -586,3 +594,3 @@ return this.asyncDispatch({
         } catch (error) {
-          if (onError) onError({
+          if (onError) await onError({
             error
@@ -610,3 +618,3 @@ });
         } catch (error) {
-          if (onError) onError({
+          if (onError) await onError({
             error
@@ -767,3 +775,7 @@ });
       emitter = null,
-      onError
+      onError,
+      eventualConsistency = {
+        isConsistent: async message => true,
+        saveMessage: async message => undefined
+      }
     }) {
@@ -804,3 +816,4 @@ if (!this.transports[transport]) {
           queueName,
-          onError
+          onError,
+          eventualConsistency
         });
@@ -811,3 +824,5 @@ channel.consume(queueName, onMessage, {
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error
+        });
         connection.close();
@@ -823,3 +838,4 @@ throw error;
       queueName,
-      onError
+      onError,
+      eventualConsistency
     }) {
@@ -836,7 +852,10 @@ return async msg => {
         try {
-          const onErrorCallback = ({
+          const onErrorCallback = async ({
             error
           }) => {
             if (!retryPolicy) channel.nack(msg, false, false);
-            if (onError) onError(error);
+            if (onError) await onError({
+              error,
+              message
+            });
             const {
@@ -867,2 +886,10 @@ maxRetries,
 
+          const isConsistent = await eventualConsistency.isConsistent(message);
+
+          if (!isConsistent) {
+            channel.ack(msg);
+            return;
+          }
+
+          eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
           emitter.emit(eventName, {
@@ -874,3 +901,6 @@ message,
         } catch (error) {
-          if (onError) onError(error);
+          if (onError) await onError({
+            error,
+            message
+          });
           channel.nack(msg, false, false);
@@ -895,3 +925,7 @@ }
       emitter = null,
-      onError
+      onError,
+      eventualConsistency = {
+        isConsistent: async message => true,
+        saveMessage: async message => undefined
+      }
     }) {
@@ -928,3 +962,4 @@ if (!this.transports[transport]) {
             emitter: domainEventsEmitter,
-            onError
+            onError,
+            eventualConsistency
           }),
@@ -937,3 +972,5 @@ options: {
       } catch (error) {
-        if (onError) onError(error);
+        if (onError) await onError({
+          error
+        });
         connection.close();
@@ -947,3 +984,4 @@ throw error;
       emitter,
-      onError
+      onError,
+      eventualConsistency
     }) {
@@ -960,8 +998,19 @@ return async msg => {
         try {
+          const isConsistent = await eventualConsistency.isConsistent(message);
+
+          if (!isConsistent) {
+            channel.ack(msg);
+            return;
+          }
+
           emitter.emit(eventName, {
             message
           });
+          eventualConsistency.saveMessage && (await eventualConsistency.saveMessage(message));
           channel.ack(msg);
         } catch (error) {
-          if (onError) onError(error);
+          if (onError) await onError({
+            error,
+            message
+          });
           console.error(`Domain event consumer error occurred: ${error.message}`);
@@ -968,0 +1017,0 @@ channel.nack(msg, false, false);

dist/utils/message.d.ts

 import { Message } from 'amqplib';
 import { message, messageBody } from '../types';
-declare const createEvent: ({ payload, meta }: message) => messageBody;
-declare const createCommand: ({ payload, meta }: message) => messageBody;
+declare const createEvent: ({ payload, meta, entityId }: message) => messageBody;
+declare const createCommand: ({ payload, meta, entityId }: message) => messageBody;
 declare const toJSON: (message: Message) => any;
 export { createEvent, createCommand, toJSON, };

package.json

 {
   "name": "@thecolvinco/nodejs-messenger",
   "description": "Message library for nodejs applications",
-  "version": "1.2.0",
+  "version": "1.4.0",
   "source": "src/main.ts",
@@ -6,0 +6,0 @@ "main": "dist/main.js",

New alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Major refactor

Supply chain risk

Package has recently undergone a major refactor. It may be unstable or indicate significant internal changes. Use caution when updating to versions that include significant changes.

Found 1 instance in 1 package

Fixed alerts

License Policy Violation

License

This package is not allowed per your license policy. Review the package's license to ensure compliance.

Found 1 instance in 1 package

Improved metrics

Total package byte prevSize

1628543

increased by394.87%

Number of package files

178

increased by368.42%

Lines of code

19913

increased by405.53%

Worsened metrics

Number of low supply chain risk alerts

10

increased by400%

Number of medium supply chain risk alerts

1

increased byInfinity%

No dependency changes