diff --git a/lib/amqp/Subscription.js b/lib/amqp/Subscription.js index 9cf3d4d..5aa7c6b 100644 --- a/lib/amqp/Subscription.js +++ b/lib/amqp/Subscription.js @@ -95,7 +95,10 @@ function Subscription(broker, vhost, subscriptionConfig, counter) { session._incrementUnacknowledgeMessageCount(message.fields.consumerTag); decorateWithRoutingHeaders(message); - if (immediateNack(message)) return ackOrNack(session, message, true); + if (immediateNack(message)) return { + debug('Immediately nacking message: %s from queue: %s', message.properties.messageId, config.queue); + ackOrNack(session, message, true); + } decorateWithRedeliveries(message, (err) => { if (err) return handleRedeliveriesError(err, session, message); @@ -210,8 +213,8 @@ function Subscription(broker, vhost, subscriptionConfig, counter) { function immediateNack(message) { const originalQueue = message.properties.headers.rascal.originalQueue; const xDeathRecords = message.properties.headers['x-death'] || []; - const currentXDeath = xDeathRecords.find(({ queue, reason }) => queue === originalQueue && reason === 'rejected') || { count: 0 }; - const previousXDeath = _.get(message, ['properties', 'headers', 'rascal', 'recovery', originalQueue, 'xDeath'], { count: 0 }); + const currentXDeath = xDeathRecords.find(({ queue, reason }) => queue === originalQueue && reason === 'rejected') || { count: 0, queue: originalQueue }; + const previousXDeath = _.get(message, ['properties', 'headers', 'rascal', 'recovery', originalQueue, 'xDeath'], { count: 0, queue: originalQueue }); const hasImmediateNackHeader = _.has(message, ['properties', 'headers', 'rascal', 'recovery', originalQueue, 'immediateNack']); if (!hasImmediateNackHeader) return false; debug('Message %s has been marked for immediate nack. Previous xDeath is %o. Current xDeath is %o.', message.properties.messageId, previousXDeath, currentXDeath);