Gestionar Jobs

Implementar queues en nuestro módulo

Una vez instalado, debemos crear las queues necesarias y crear tanto los jobs como los consumidores de cada queue.

Dentro del fichero src/app.queues.ts tendremos un objeto donde definiremos cada queue, asociaremos cada queue a cada módulo desde el cual se enviará o consumirá.

import { QueueRegister } from '@app/queue-manager/queue-manager.types';

// storages where all jobs are stored
export enum QueueStorage {
    COMMON_MAIL = 'commonMail',
}

// relationship between storage and bounded context
export const appQueues: QueueRegister = {
    // common is for all bounded contexts
    common: [
        {
            name: QueueStorage.COMMON_MAIL,
        },
    ],
    iam: [
        {
            name: QueueStorage.COMMON_MAIL,
        },
    ],
};

Dentro de la variable QueueStorage definiremos un identificativo único para cada queue, en este ejemplo COMMON_MAIL = 'mail'.

A continuación debemos importar en cada módulo que gestione queues, el módulo dinámico BullModule.registerQueue estableciendo el identificador common y el suyo propio.

BullModule.registerQueue(
    ...appQueues.common,
    ...appQueues.iam,
),

De esta manera estaríamos preparados para empezar a lazar jobs y definir su consumidores.

¡Recuerda! todos los queues que desee usar en un módulo, deben de estar previamente registrados en la definición del módulo dentro de BullModule.registerQueue

Definir un consumer

Para procesar los jobs enviados a cada cola debemos implementar un consumer que procese la tarea esperada, para esto necesitaremos crear un consumer, y declararlo, este sería un ejemplo de consumer.

import { Process, Processor } from '@nestjs/bull';
import { Job } from 'bull';
import { QueueStorage } from 'src/app.queues';

@Processor(QueueStorage.COMMON_MAIL)
export class CommonMailConsumer
{
    @Process('sendMail')
    async main(job: Job<{
        payload: any;
        otherData: any;
        otherData1: any;
    }>): Promise<void>
    {
        const { payload, otherData, otherData1 } = job.data;
        // A partir e este punto procesarías los datos recibidos
    }
}

Posteriormente deberemos declarar el consumer en su módulo correspondiente, en el array de providers, como si un servicio se tratara.

Para más información sobre consumers podéis acceder a nestjs consumers.

Por convención los consumers los definimos en:

src/@api/my-bounded-context/my-module/shared/consumers/

Definir un job

Para definir un job en una queue necesitaremos inyectar el QueueManagerJobService y la Queue donde publicar el job.

constructor(
    private readonly jobService: QueueManagerJobService,
    @InjectQueue(QueueStorage.COMMON_MAIL) private commonMailQueue: Queue,
) {}

QueueManagerJobService se importa desde @api/queue-manager/job/shared/services/job.service y Queue desde bull.

En la inyección de Queue usaremos la clave única definida en el QueueStorage para obtener la instancia de la queue deseada.

Una vez importado usaremos el método add para lanzar el job a la queue correspondiente.

handleQueue(): void
{
    this.jobService.add(
        this.commonMailQueue,
        {
            foo: 'bar',
        },
        ['common-mail', '[e-mail]'],
    );
}

Los parámetros que pasamos es la queue donde vamos a definir:

• La queue donde vamos a definir el job.

• El payload que requiere el job.

• String o array de string para crear unos tags que nos sirvan para localizar ese job en concreto.

Adicionalmente podemos añadir opts como parámetro opciones para el job, consultar JobOpts para más información, o un jobName para organizar los jobs dentro de una queue, consultar Named Jobs para más información.

Use la ayuda de Typescript para consultar los diferentes formas de llamar al método add.