概要

今回はNestJS上でBullを使ってみたサンプルです。

docs.nestjs.com

BullとはRedisベースのメッセージキューです。 github.com

実行環境

Node.js - 10.x
redis - 5.x

使用ライブラリ

@nestjs/core - 6.7.x
@nestjs/bull - 0.0.1
bull - 3.12.1
@types/bull - 3.12.x

サンプルソース

app.processor.ts

Processorデコレータに test を定義します。
処理は5秒待機してコンソールログを出力するものです。

import { Process, Processor } from '@nestjs/bull';
import { Logger } from '@nestjs/common';
import { Job } from 'bull';

@Processor('test')
export class AppProcessor {
  @Process()
  async exec(job: Job<unknown>) {
    await new Promise((resolve) => setTimeout(resolve, 5000));
    Logger.debug(`id=${job.id}, timestamp=${job.timestamp}`);
  }
}

app.service.ts

InjectQueue に test を指定します。
Queue.add でキューにデータを登録します。
登録したジョブIDを戻します。

import { InjectQueue } from '@nestjs/bull';
import { Injectable } from '@nestjs/common';
import { Queue } from 'bull';

@Injectable()
export class AppService {
  constructor(@InjectQueue('test') private readonly testQueue: Queue) {}

  async addQueue() {
    const job = await this.testQueue.add({}, { lifo: true });
    return job.id;
  }
}

app.controller.ts

import { Controller, Get } from '@nestjs/common';
import { AppService } from './app.service';

@Controller()
export class AppController {
  constructor(private readonly appService: AppService) {}

  @Get('bull_test')
  bullTest() {
    return this.appService.addQueue();
  }
}

app.module.ts

BullModuleを定義します。定義時に test を定義します。

import { BullModule } from '@nestjs/bull';
import { Module } from '@nestjs/common';

import { AppController } from './app.controller';
import { AppProcessor } from './app.processor';
import { AppService } from './app.service';

@Module({
  imports: [
    BullModule.registerQueue({
      name: 'test',
      redis: {
        host: 'localhost',
        port: 6379,
      },
    }),
  ],
  controllers: [AppController],
  providers: [AppService, AppProcessor],
})
export class AppModule {}