fix queue and exchange assertions

src/Publisher.js

@@ -13,7 +13,13 @@ class Publisher {
     this.exchange = exchange
     this.routingKey = ''
     this.options = options
-    const { MessageModel, ContentSchema } = options || {}
+    const { MessageModel, ContentSchema, assertExchange = true } = options || {}
+    this._assertExchange = null
+    if (assertExchange) {
+      this._assertExchange = assertExchange === true
+        ? { durable: true } // defaults
+        : assertExchange
+    }
     this.MessageModel = MessageModel || QueueMessage
     this.ContentSchema = ContentSchema || JSON
   }
@@ -25,7 +31,9 @@ class Publisher {
    * @returns {Promise}
    */
   assertExchangeOrQueue (channel) {
-    return channel.assertExchange(this.exchange, 'fanout', { durable: true })
+    if (this._assertExchange) {
+      return channel.assertExchange(this.exchange, 'fanout', this._assertExchange)
+    }
   }
 
   async initialize () {

src/QueueClient.js

@@ -10,14 +10,26 @@ class QueueClient extends Publisher {
   constructor (queueConnection, logger, name, options) {
     super(queueConnection, logger, '', options)
     this.routingKey = name
+    this._assertQueue = null
+    const {
+      assertQueue
+    } = options || {}
+
+    if (assertQueue) {
+      this._assertQueue = assertQueue === true
+        ? { durable: true } // defaults
+        : assertQueue
+    }
   }
 
   /**
    * @param channel
    * @returns {Promise}
    */
   assertExchangeOrQueue (channel) {
-    return channel.assertQueue(this.routingKey, { durable: true })
+    if (this._assertQueue) {
+      return channel.assertQueue(this.routingKey, this._assertQueue)
+    }
   }
 }
 

src/QueueServer.js

@@ -10,18 +10,17 @@ class QueueServer extends Subscriber {
   constructor (queueConnection, logger, name, options) {
     super(queueConnection, logger, name, options)
     const {
-      assertQueueOptions = true,
+      assertQueue = true,
       prefetchCount
     } = options || {}
 
-    this._assertQueueOptions = null
+    this._assertQueue = null
     this._prefetchCount = prefetchCount
 
-    if (assertQueueOptions) {
-      const defaultOptions = { durable: true }
-      this._assertQueueOptions = assertQueueOptions === true
-        ? defaultOptions
-        : assertQueueOptions
+    if (assertQueue) {
+      this._assertQueue = assertQueue === true
+        ? { durable: true } // defaults
+        : assertQueue
     }
   }
 
@@ -31,8 +30,8 @@ class QueueServer extends Subscriber {
   async initialize () {
     try {
       const channel = await this._connection.getChannel()
-      if (this._assertQueueOptions) {
-        await channel.assertQueue(this.name, this._assertQueueOptions)
+      if (this._assertQueue) {
+        await channel.assertQueue(this.name, this._assertQueue)
       }
       await channel.prefetch(this._prefetchCount)
       await channel.consume(this.name, (msg) => {

src/RPCClient.js

@@ -21,21 +21,20 @@ class RPCClient {
       RequestContentSchema,
       ResponseContentSchema,
       replyQueueName = '',
-      replyQueueOptions = true
+      assertReplyQueue = true
     } = options || {}
 
     this._connection = queueConnection
     this._logger = logger
     this.name = rpcName
     this._replyQueue = replyQueueName || ''
-    this._replyQueueOptions = null
+    this._assertReplyQueue = null
     this._correlationIdMap = new Map()
 
-    if (replyQueueOptions) {
-      const defaultOptions = { exclusive: true }
-      this._replyQueueOptions = replyQueueOptions === true
-        ? defaultOptions
-        : replyQueueOptions
+    if (assertReplyQueue) {
+      this._assertReplyQueue = assertReplyQueue === true
+        ? { exclusive: true } // defaults
+        : assertReplyQueue
     }
 
     this._rpcQueueMaxSize = queueMaxSize
@@ -157,8 +156,8 @@ class RPCClient {
    * */
   async _getReplyQueue (ch) {
     try {
-      if (this._replyQueueOptions) {
-        const assertResult = await ch.assertQueue(this._replyQueue, this._replyQueueOptions)
+      if (this._assertReplyQueue) {
+        const assertResult = await ch.assertQueue(this._replyQueue, this._assertReplyQueue)
         this._replyQueue = assertResult.queue
       }
 

src/RPCServer.js

@@ -19,19 +19,18 @@ class RPCServer {
       ResponseMessageModel,
       RequestContentSchema,
       ResponseContentSchema,
-      queueOptions = true
+      assertQueue = true
     } = options || {}
 
     this._connection = queueConnection
     this._logger = logger
     this.name = rpcName
-    this._queueOptions = null
+    this._assertQueue = null
 
-    if (queueOptions) {
-      const defaultOptions = { durable: true }
-      this._queueOptions = queueOptions === true
-        ? defaultOptions
-        : queueOptions
+    if (assertQueue) {
+      this._assertQueue = assertQueue === true
+        ? { durable: true } // defaults
+        : assertQueue
     }
 
     this._prefetchCount = prefetchCount
@@ -82,8 +81,8 @@ class RPCServer {
   async initialize () {
     try {
       const channel = await this._connection.getChannel()
-      if (this._queueOptions) {
-        await channel.assertQueue(this.name, this._queueOptions)
+      if (this._assertQueue) {
+        await channel.assertQueue(this.name, this._assertQueue)
       }
       await channel.prefetch(this._prefetchCount)
       await channel.consume(this.name, (msg) => {

src/Subscriber.js

@@ -12,11 +12,27 @@ class Subscriber {
     this._logger = logger
     this.name = name
 
-    const { maxRetry, timeoutMs, MessageModel, ContentSchema } = options || {}
+    const {
+      maxRetry,
+      timeoutMs,
+      MessageModel,
+      ContentSchema,
+      assertQueueOptions,
+      assertExchange = true
+    } = options || {}
     this._maxRetry = maxRetry
     this._timeoutMs = timeoutMs
     this.MessageModel = MessageModel || QueueMessage
     this.ContentSchema = ContentSchema || JSON
+    this._assertQueueOptions = assertQueueOptions
+      ? Object.assign(assertQueueOptions || {}, { exclusive: true })
+      : { exclusive: true } // defaults
+    this._assertExchange = null
+    if (assertExchange) {
+      this._assertExchange = assertExchange === true
+        ? { durable: true } // defaults
+        : assertExchange
+    }
 
     this._retryMap = new Map()
 
@@ -53,8 +69,10 @@ class Subscriber {
   async initialize () {
     try {
       const channel = await this._connection.getChannel()
-      await channel.assertExchange(this.name, 'fanout', { durable: true })
-      const queue = await channel.assertQueue('', { exclusive: true })
+      if (this._assertExchange) {
+        await channel.assertExchange(this.name, 'fanout', this._assertExchange)
+      }
+      const queue = await channel.assertQueue('', this._assertQueueOptions)
 
       await channel.bindQueue(queue.queue, this.name, '')
 
@@ -113,16 +131,11 @@ class Subscriber {
   /**
    * @param channel
    * @param msg
-   * @return {Promise}
+   * @param request
+   * @returns {boolean} true if too many retries reached
    * @private
    */
-  async _processMessage (channel, msg) {
-    const request = this._parseMessage(msg)
-    if (!request) {
-      this._ack(channel, msg)
-      return
-    }
-
+  _handleMessageRetry (channel, msg, request) {
     if (msg.fields && msg.fields.redelivered && msg.fields.consumerTag) {
       let counter = 1
       if (this._retryMap.has(msg.fields.consumerTag)) {
@@ -138,10 +151,31 @@ class Subscriber {
         if (msg.fields.consumerTag) {
           this._retryMap.delete(msg.fields.consumerTag)
         }
-        return
+        return true
       }
     }
 
+    return false
+  }
+
+  /**
+   * @param channel
+   * @param msg
+   * @return {Promise}
+   * @private
+   */
+  async _processMessage (channel, msg) {
+    const request = this._parseMessage(msg)
+    if (!request) {
+      this._ack(channel, msg)
+      return
+    }
+
+    const tooManyRetries = this._handleMessageRetry(channel, msg, request)
+    if (tooManyRetries) {
+      return
+    }
+
     let timedOut = false
     const timeoutMs = typeof request.timeOut === 'number' ? request.timeOut : this._timeoutMs
     const timer = setTimeout(() => {
@@ -150,23 +184,23 @@ class Subscriber {
       this._nack(channel, msg)
     }, timeoutMs)
 
-    return Promise.resolve().then(() => {
-      return this._callback(request.data, msg.properties, request, msg)
-    }).then(() => {
+    try {
+      await this._callback(request.data, msg.properties, request, msg)
+
       if (!timedOut) {
         clearTimeout(timer)
         this._ack(channel, msg)
         if (msg.fields && msg.fields.redelivered && msg.fields.consumerTag) {
           this._retryMap.delete(msg.fields.consumerTag)
         }
       }
-    }).catch((err) => {
+    } catch (err) {
       if (!timedOut) {
         clearTimeout(timer)
         this._logger.error('Cannot process Subscriber consume', err)
         this._nack(channel, msg)
       }
-    })
+    }
   }
 
   /**