Merge branch 'issue-86' into kafka-client

Author: AndreMaz

src/adapters/amqp.js

@@ -93,9 +93,6 @@ class AmqpAdapter extends BaseAdapter {
 				.map(s => s.trim());
 		}
 
-		this.error2ErrorInfoParser =
-			this.opts?.deadLettering?.error2ErrorInfoParser || error2ErrorInfoParser;
-
 		/** @type {AMQPLibConnection} */
 		this.connection = null;
 		/** @type {AMQPLibChannel} */

src/adapters/base.js

@@ -11,6 +11,7 @@ const semver = require("semver");
 const { MoleculerError } = require("moleculer").Errors;
 const { Serializers, METRIC } = require("moleculer");
 const C = require("../constants");
+const { error2ErrorInfoParser } = require("../utils");
 
 /**
  * @typedef {import("moleculer").ServiceBroker} ServiceBroker Moleculer Service Broker instance
@@ -58,6 +59,14 @@ class BaseAdapter {
 
 		/** @type {Boolean} Flag indicating the adapter's connection status */
 		this.connected = false;
+
+		/**
+		 * Function to convert Error to a plain object and gets error info ready to be placed in message headers
+		 *
+		 * @type {(error: Error) => Record<string, string>}
+		 */
+		this.error2ErrorInfoParser =
+			this.opts?.deadLettering?.error2ErrorInfoParser || error2ErrorInfoParser;
 	}
 
 	/**

src/adapters/kafka.js

@@ -11,7 +11,6 @@ const _ = require("lodash");
 const { MoleculerError, MoleculerRetryableError } = require("moleculer").Errors;
 const C = require("../constants");
 const { INVALID_MESSAGE_SERIALIZATION_ERROR_CODE } = require("../constants");
-const { error2ErrorInfoParser } = require("../utils");
 /** Name of the partition where an error occurred while processing the message */
 const HEADER_ORIGINAL_PARTITION = "x-original-partition";
 
@@ -70,9 +69,6 @@ class KafkaAdapter extends BaseAdapter {
 			}
 		});
 
-		this.error2ErrorInfoParser =
-			this.opts?.deadLettering?.error2ErrorInfoParser || error2ErrorInfoParser;
-
 		/** @type {KafkaConsumer} */
 		this.client = null;
 

src/adapters/nats.js

@@ -82,9 +82,6 @@ class NatsAdapter extends BaseAdapter {
 			}
 		});
 
-		this.error2ErrorInfoParser =
-			this.opts?.deadLettering?.error2ErrorInfoParser || error2ErrorInfoParser;
-
 		// Adapted from: https://github.com/moleculerjs/moleculer/blob/3f7e712a8ce31087c7d333ad9dbaf63617c8497b/src/transporters/nats.js#L141-L143
 		if (this.opts.nats.url)
 			this.opts.nats.connectionOptions.servers = this.opts.nats.url

src/adapters/redis.js

@@ -101,9 +101,8 @@ class RedisAdapter extends BaseAdapter {
 			}
 		});
 
+		/** @type {Number} Time-to-live in seconds for error info storage (only for Redis adapter) */
 		this.errorInfoTTL = this.opts?.deadLettering?.errorInfoTTL || 24 * 3600; // 24 hours
-		this.error2ErrorInfoParser =
-			this.opts?.deadLettering?.error2ErrorInfoParser || error2ErrorInfoParser;
 
 		/**
 		 * @type {Map<string,Cluster|Redis>}

src/index.js

@@ -27,7 +27,7 @@ const C = require("./constants");
  * @property {String} exchangeName Name of the dead-letter exchange (only for AMQP adapter)
  * @property {Object} exchangeOptions Options for the dead-letter exchange (only for AMQP adapter)
  * @property {Object} queueOptions Options for the dead-letter queue (only for AMQP adapter)
- * @property {(error: Error) => Record<string, any>} [error2ErrorInfoParser] Function to convert Error object to a plain object
+ * @property {(error: Error) => Record<string, string>} [error2ErrorInfoParser] Function to convert Error object to a plain object
  * @property {Number} errorInfoTTL Time-to-live in seconds for error info storage (only for Redis adapter)
  */