tpz-queue

Módulo dedicado a publicar e consumir mensagens das diversas filas RabbitMQ utilizadas pela Centricx.

Instalação

npm install apx-queue -P

Filas

Cada objeto exposto por este módulo representa uma interface padronizada para utilizar a fila. Para utilizar o módulo:

const { transferConfirm } = require('apx-queue');

const message = { author: 'Karl Marx', title: 'Das Kapital' };
transferConfirm.publish(message);

As filas disponíveis são:

| Nome | API | Finalidade | | --- | --- | --- | | centricxSurveyRequest | survey | Envio de dados das pesquisas para um base de dados PostgreSql |

Métodos

Cada fila possui os seguintes métodos:

queue.publish(message, headers)

Retorna Promise que resolve true para sucesso, false para erro. (detalhes metodo publish no pacote amqplib)

Publica mensagem na exchange e routingKey configurados na fila.

message e headers são objetos JSON.

queue.sendToQueue(message, headers)

Retorna Promise que resolve true para sucesso, false para erro. (detalhes metodo sendToQueue no pacote amqplib)

Publicar mensagem diretamente na fila (sem passar por exchange).

message e headers são objetos JSON.

queue.consume(consumer)

Retorna a própria queue para permitir chain de outros métodos:

queue
  .consume(consumer)
  .retry([
    5 * seconds,
    10 * seconds,
    20 * seconds
  ])
  .catch(errorHandler);

Registrar um consumidor para as mensagens da fila.

consumer é uma função com a seguinte estrutura:

const consumer = (content, props, fail, retry) => {
  const attempts = props.headers['X-attempts'];
  const date = new Date();

  console.log(date, '-- attempts', attempts, '-->', content);

  retry(new Error(`attempt ${attempts}, Im not in the mood`));
};

content e props são objetos com o corpo e os metadados da mensagem, enquanto fail e retry são callbacks para executar em caso de exceção. fail vai executar o código registrado no método queue.catch. retry vai enviar a mensagem para uma fila de retry, sendo reenviada para a fila de trabalho no tempo determinado no método queue.retry.

queue.retry(timesList)

Retorna a própria queue para permitir chain de outros métodos.

timeLists é um array de milisegundos utilizados para criar as filas de retry associados via DLX com a fila de trabalho (queue). Cada item do array corresponde a uma fila configurada com o TTL especificado.

A mensagem é enviada para a fila correspondente pelo consumidor (usando o callback retry), incrementando o header x-attempts. A primeira retentativa vai enviar a mensagem para a primeira fila de retry, a segunda retentativa para a segunda fila de retry. Quando se esgotarem as filas de retentativa o errorHandler registrado no método queue.catch será executado com o erro informado.

queue.catch(errorHandler)

Retorna a própria queue.

Registra o errorHandler para lidar com erros disparados pelo consumidor, tanto pelo callback fail, quando pelo callback retry quando o número de retentativas se esgotarem.