fix: use optional access to message headers

Author: mciuchitu

package.json

@@ -90,13 +90,13 @@
         "@nestjs/cqrs": "^9.0.1",
         "@nestjs/testing": "^9.0.4",
         "@release-it/conventional-changelog": "^5.0.0",
-        "@types/amqplib": "0.8.2",
+        "@types/amqplib": "^0.8.2",
         "@types/jest": "^28.1.6",
         "@types/lodash": "^4.14.182",
         "@types/minimist": "^1.2.2",
         "@types/semver": "^7.3.10",
-        "amqp-connection-manager": "^4.1.3",
-        "amqplib": "^0.10.0",
+        "amqp-connection-manager": "^4.1.6",
+        "amqplib": "^0.10.2",
         "commitizen": "^4.2.5",
         "cz-conventional-changelog": "^3.3.0",
         "eslint-plugin-jest": "^26.6.0",

src/interface/AbstractSubscriptionEvent.ts

@@ -16,6 +16,8 @@ export abstract class AbstractSubscriptionEvent<T extends Record<string, any> |
     };
 
     get retryCount(): number {
-        return this.#message?.properties.headers[RETRY_COUNT_HEADER] ?? 0;
+        // incorrect typing from @types/amqplib
+        // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+        return this.#message?.properties.headers?.[RETRY_COUNT_HEADER] ?? 0;
     }
 }

src/service/PubSubEventBinder.ts

@@ -6,9 +6,8 @@ import type { IPubsubEventOptions } from '../decorator';
 import { PubsubEvent, PubsubEventHandler } from '../decorator';
 import type { AbstractPubsubAnyEventHandler, AbstractSubscriptionEvent, IEventWrapper, IHandlerWrapper, IPubsubEventHandlerMetadata } from '../interface';
 import { LoggerProvider } from '../provider';
-import { generateQueueName, toEventName } from '../utils';
+import { generateQueueName, getMessageExchange, toEventName } from '../utils';
 import { FAN_OUT_BINDING } from '../utils/configuration';
-import { ORIGIN_EXCHANGE_HEADER } from '../utils/retry-constants';
 import { Consumer } from './Consumer';
 import { EventBus } from './EventBus';
 import { PubSubReflector } from './PubSubReflector';
@@ -58,7 +57,7 @@ export class PubSubEventBinder {
             this.consumer.ack(message);
             return;
         }
-        const messageExchange = message.properties.headers[ORIGIN_EXCHANGE_HEADER] ?? message.fields.exchange;
+        const messageExchange = getMessageExchange(message);
 
         // try exact match at first
         let matchedEventWrappers: IEventWrapper[] = eventWrappers.filter(

src/service/retry-strategy/DeadLetterTtlRetryStrategy.ts

@@ -14,7 +14,7 @@ import {
     ORIGIN_EXCHANGE_HEADER,
     RETRY_COUNT_HEADER,
 } from '../../utils/retry-constants';
-import { calculateDelay } from '../../utils';
+import { calculateDelay, getMessageExchange } from '../../utils';
 import type { IRetryStrategy } from './IRetryStrategy';
 
 @Injectable()
@@ -94,7 +94,7 @@ export class DeadLetterTtlRetryStrategy implements IRetryStrategy {
             headers: {
                 ...message.properties.headers,
                 [RETRY_COUNT_HEADER]: retryCount,
-                [ORIGIN_EXCHANGE_HEADER]: message.properties.headers[ORIGIN_EXCHANGE_HEADER] ?? message.fields.exchange,
+                [ORIGIN_EXCHANGE_HEADER]: getMessageExchange(message),
             },
         });
 

src/service/retry-strategy/DelayedMessageExchangeRetryStrategy.ts

@@ -7,7 +7,7 @@ import { DefaultedRetryOptions, ExchangeOptions, RetryStrategyEnum } from '../..
 import { LoggerProvider } from '../../provider';
 import { CQRS_EXCHANGE_CONFIG, CQRS_RETRY_OPTIONS } from '../../utils/configuration';
 import { DEFAULT_RETRY_DELAYED_MESSAGE_EXCHANGE_NAME, ORIGIN_EXCHANGE_HEADER, RETRY_COUNT_HEADER } from '../../utils/retry-constants';
-import { calculateDelay } from '../../utils';
+import { calculateDelay, getMessageExchange } from '../../utils';
 import type { IRetryStrategy } from './IRetryStrategy';
 
 @Injectable()
@@ -66,7 +66,7 @@ export class DelayedMessageExchangeRetryStrategy implements IRetryStrategy {
                 ...message.properties.headers,
                 'x-delay': delayValue,
                 [RETRY_COUNT_HEADER]: retryCount,
-                [ORIGIN_EXCHANGE_HEADER]: message.properties.headers[ORIGIN_EXCHANGE_HEADER] ?? message.fields.exchange,
+                [ORIGIN_EXCHANGE_HEADER]: getMessageExchange(message),
             },
         });
 

src/utils/helpers.ts

@@ -1,7 +1,9 @@
 import type { Type } from '@nestjs/common';
 import snakeCase from 'lodash/snakeCase';
 import startCase from 'lodash/startCase';
+import type { Message } from 'amqplib';
 import type { AbstractPubsubAnyEventHandler, DelayType } from '../interface';
+import { ORIGIN_EXCHANGE_HEADER } from './retry-constants';
 
 /**
  * Transform an event string (event class name) to a RabbitMQ event.
@@ -53,3 +55,9 @@ export function generateQueueName(handler: Type<AbstractPubsubAnyEventHandler>):
 
     return [platform, toSnakeCase(handler.name)].join(':');
 }
+
+export function getMessageExchange(message: Message): string {
+    // incorrect typing from @types/amqplib
+    // eslint-disable-next-line @typescript-eslint/no-unnecessary-condition
+    return message.properties.headers?.[ORIGIN_EXCHANGE_HEADER] ?? message.fields.exchange;
+}

tsconfig.build-cjs.json

@@ -1,8 +1,8 @@
 {
     "extends": "./tsconfig.build.json",
     "compilerOptions": {
-        "target": "ES2018",
-        "lib": ["ES2018"],
+        "target": "ES2019",
+        "lib": ["ES2019"],
         "module": "commonjs",
         "outDir": "dist/cjs"
     }