RabbitMQ retries - send failed messages to retry loop

amqp-send-to-retry.js

// emulate some job which can be failed
function handler() {
  return Math.random() > 0.1 ? Promise.resolve() : Promise.reject()
}

channel.consume('comments', msg => {
  return (new Promise((resolve, reject) => {
    if (msg.fields.redelivered) {
      reject('Message was redelivered, so something wrong happened');
      return;
    }

    handler(msg, channel)
      .then(resolve)
      .catch(reject);
  }))
    .then(() => {
      channel.ack(msg);
    })
    .catch(handleRejectedMsg);

  function handleRejectedMsg(reasonOfFail) {
    return sendMsgToRetry({ msg, queue:'comments', channel, reasonOfFail });
  }
}

function sendMsgToRetry(args) {
  const channel = args.channel;
  const queue = args.queue;
  const msg = args.msg;

  // ack original msg
  channel.ack(msg);

  // Unpack content, update and pack it back
  function getAttemptAndUpdatedContent(msg) {
    let content = JSON.parse(msg.content.toString(contentEncoding));
    content.try_attempt = ++content.try_attempt || 1;
    const attempt = content.try_attempt;
    content = Buffer.from(JSON.stringify(content), contentEncoding);

    return { attempt, content };
  }

  const { attempt, content } = getAttemptAndUpdatedContent(msg);

  if (attempt <= 3) {
    const routingKey = `retry-${attempt}`;
    const options = {
      contentEncoding,
      contentType: contentTypeJson,
      persistent: true,
    };

    Object.keys(msg.properties).forEach(key => {
      options[key] = msg.properties[key];
    });

    return channel.publish('TTL-comments', routingKey, content, options);
  }

  return Promise.resolve();
}