src/gc.js
function collectMessage(message, processingQueue, cb) {
        const threshold = message.getRetryThreshold();
        const messageRetryThreshold = typeof threshold === 'number' ? threshold : dispatcher.getMessageRetryThreshold();

        const delay = message.getRetryDelay();
        const messageRetryDelay = typeof delay === 'number' ? delay : dispatcher.getMessageRetryDelay();

        let destQueueName = null;
        let delayed = false;
        let requeued = false;

        const multi = client.multi();

        /**
         * Only exceptions from non periodic messages are handled.
         * Periodic messages are ignored once they are delivered to a consumer.
         */
        if (!dispatcher.isPeriodic(message)) {
            const uuid = message.getId();

            /**
             * Attempts
             */
            const attempts = increaseAttempts(message);

            /**
             *
             */
            if (attempts < messageRetryThreshold) {
                debug(`Trying to consume message ID [${uuid}] again (attempts: [${attempts}]) ...`);
                if (messageRetryDelay) {
                    debug(`Scheduling message ID [${uuid}]  (delay: [${messageRetryDelay}])...`);
                    message.setScheduledDelay(messageRetryDelay);
                    dispatcher.schedule(message, multi);
                    delayed = true;
                } else {
                    debug(`Message ID [${uuid}] is going to be enqueued immediately...`);
                    destQueueName = keyQueueName;
                    requeued = true;
                }
            } else {
                debug(`Message ID [${uuid}] has exceeded max retry threshold...`);
                destQueueName = keyQueueNameDead;
            }

            /**
             *
             */
            if (destQueueName) {
                debug(`Moving message [${uuid}] to queue [${destQueueName}]...`);
                multi.lpush(destQueueName, message.toString());
            }
            multi.rpop(processingQueue);
            multi.exec((err) => {
                if (err) cb(err);
                else {
                    if (dispatcher.isTest()) {
                        if (requeued) dispatcher.emit(events.MESSAGE_REQUEUED, message);
                        else if (delayed) dispatcher.emit(events.MESSAGE_DELAYED, message);
                        else dispatcher.emit(events.MESSAGE_DEAD_LETTER, message);
                    }
                    cb();
                }
            });
        } else {
            client.rpop(processingQueue, cb);
        }
    }
function collectMessage(message, processingQueue, cb) {
        const threshold = message.getRetryThreshold();
        const messageRetryThreshold = typeof threshold === 'number' ? threshold : dispatcher.getMessageRetryThreshold();

        const delay = message.getRetryDelay();
        const messageRetryDelay = typeof delay === 'number' ? delay : dispatcher.getMessageRetryDelay();

        let destQueueName = null;
        let delayed = false;
        let requeued = false;

        const multi = client.multi();

        /**
         * Only exceptions from non periodic messages are handled.
         * Periodic messages are ignored once they are delivered to a consumer.
         */
        if (!dispatcher.isPeriodic(message)) {
            const uuid = message.getId();

            /**
             * Attempts
             */
            const attempts = increaseAttempts(message);

            /**
             *
             */
            if (attempts < messageRetryThreshold) {
                debug(`Trying to consume message ID [${uuid}] again (attempts: [${attempts}]) ...`);
                if (messageRetryDelay) {
                    debug(`Scheduling message ID [${uuid}]  (delay: [${messageRetryDelay}])...`);
                    message.setScheduledDelay(messageRetryDelay);
                    dispatcher.schedule(message, multi);
                    delayed = true;
                } else {
                    debug(`Message ID [${uuid}] is going to be enqueued immediately...`);
                    destQueueName = keyQueueName;
                    requeued = true;
                }
            } else {
                debug(`Message ID [${uuid}] has exceeded max retry threshold...`);
                destQueueName = keyQueueNameDead;
            }

            /**
             *
             */
            if (destQueueName) {
                debug(`Moving message [${uuid}] to queue [${destQueueName}]...`);
                multi.lpush(destQueueName, message.toString());
            }
            multi.rpop(processingQueue);
            multi.exec((err) => {
                if (err) cb(err);
                else {
                    if (dispatcher.isTest()) {
                        if (requeued) dispatcher.emit(events.MESSAGE_REQUEUED, message);
                        else if (delayed) dispatcher.emit(events.MESSAGE_DELAYED, message);
                        else dispatcher.emit(events.MESSAGE_DEAD_LETTER, message);
                    }
                    cb();
                }
            });
        } else {
            client.rpop(processingQueue, cb);
        }
    }
function collectMessage ( message , processingQueue , cb ) { const threshold = message . getRetryThreshold ( ) ; const messageRetryThreshold = typeof threshold === 'number' ? threshold : dispatcher . getMessageRetryThreshold ( ) ; const delay = message . getRetryDelay ( ) ; const messageRetryDelay = typeof delay === 'number' ? delay : dispatcher . getMessageRetryDelay ( ) ; let destQueueName = null ; let delayed = false ; let requeued = false ; const multi = client . multi ( ) ; /**
         * Only exceptions from non periodic messages are handled.
         * Periodic messages are ignored once they are delivered to a consumer.
         */ if ( ! dispatcher . isPeriodic ( message ) ) { const uuid = message . getId ( ) ; /**
             * Attempts
             */ const attempts = increaseAttempts ( message ) ; /**
             *
             */ if ( attempts < messageRetryThreshold ) { debug ( ` ${ uuid } ${ attempts } ` ) ; if ( messageRetryDelay ) { debug ( ` ${ uuid } ${ messageRetryDelay } ` ) ; message . setScheduledDelay ( messageRetryDelay ) ; dispatcher . schedule ( message , multi ) ; delayed = true ; } else { debug ( ` ${ uuid } ` ) ; destQueueName = keyQueueName ; requeued = true ; } } else { debug ( ` ${ uuid } ` ) ; destQueueName = keyQueueNameDead ; } /**
             *
             */ if ( destQueueName ) { debug ( ` ${ uuid } ${ destQueueName } ` ) ; multi . lpush ( destQueueName , message . toString ( ) ) ; } multi . rpop ( processingQueue ) ; multi . exec ( ( err ) => { if ( err ) cb ( err ) ; else { if ( dispatcher . isTest ( ) ) { if ( requeued ) dispatcher . emit ( events . MESSAGE_REQUEUED , message ) ; else if ( delayed ) dispatcher . emit ( events . MESSAGE_DELAYED , message ) ; else dispatcher . emit ( events . MESSAGE_DEAD_LETTER , message ) ; } cb ( ) ; } } ) ; } else { client . rpop ( processingQueue , cb ) ; } }
Move message to dead-letter queue when max attempts threshold is reached
otherwise requeue it again

@param {object} message
@param {string} processingQueue
@param {function} cb
Move message to dead - letter queue when max attempts threshold is reached otherwise requeue it again